概要

在 go 中,一个 goroutine 是无法直接强制中断 另一个 goroutine 的。但是可以 goroutine A 向 goroutine B 发送请求终止信息,goroutine B 收到终止信息后中断自己,从而达到控制 goroutine 终止的目的。

dragonboat 是一个 Raft 库,其中使用 stopper 对 终止 goroutine 进行了封装。本文会给出 stopper 的使用方法和实现解析。

使用

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
package main

import (
"fmt"
"time"

"github.com/lni/goutils/syncutil"
)

func main() {
stoper := syncutil.NewStopper()
stoper.RunWorker(func() {
ticker := time.NewTicker(time.Second)
for {
select {
case <-ticker.C:
fmt.Println("Hello")
case <-stoper.ShouldStop():
fmt.Println("Closed")
return
}
}
})
time.Sleep(time.Second * 3)
stoper.Stop()
}

输出是

1
2
3
4
Hello
Hello
Hello
Closed

其中 stoper.RunWorker 执行了一个 异步 goroutine,在 stoper.ShouldStop() 收到信息前,每秒打印一次 Hello。主 goroutine 则因为 Sleep 睡了 3 秒。3 秒后 主 goroutine 睡醒,调用 stoper.Stop() 请求关闭 子 goroutine,并阻塞直到所有子 goroutine 退出。子 goroutine 收到终止信息后打印 Closed 并终止,主 goroutine 退出,程序结束。

实现

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
// Copyright 2014 The Cockroach Authors.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
// implied. See the License for the specific language governing
// permissions and limitations under the License.
//
//
//
// Copyright 2017-2019 Lei Ni (nilei81@gmail.com) and other Dragonboat authors.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

package syncutil

import (
"log"
"sync"

"github.com/lni/goutils/envutil"
"github.com/lni/goutils/lang"
)

// Stopper is a manager struct for managing worker goroutines. It is modified
// from an early version of the stopper struct found in CockroachDB's codebase.
type Stopper struct {
mu sync.Mutex
shouldStopC chan struct{}
wg sync.WaitGroup
debug bool
}

// NewStopper return a new Stopper instance.
func NewStopper() *Stopper {
s := &Stopper{
shouldStopC: make(chan struct{}),
debug: envutil.GetBoolEnvVarOrDefault("LEAKTEST", false), // 用于决定是否打印调试信息
}

return s
}

// RunWorker creates a new goroutine and invoke the f func in that new
// worker goroutine.
func (s *Stopper) RunWorker(f func()) {
s.mu.Lock()
defer s.mu.Unlock()
if s.stopped() {
// 已经被关闭了,不再添加新的子 goroutine
return
}
s.runWorker(f, "")
}

func (s *Stopper) runWorker(f func(), name string) {
s.wg.Add(1) // 等待执行,数量 ++
var gid uint64

go func() { // runWorker 内部会根据传入的 f 创建 goroutine
if s.debug {
gid = lang.GetGIDForDebugOnly()
log.Printf("goroutine %d started, name %s", gid, name)
}
f()
s.wg.Done() // 执行完毕,数量 --
if s.debug {
log.Printf("goroutine %d stopped, name %s", gid, name)
}
}()
}

// ShouldStop returns a chan struct{} used for indicating whether the
// Stop() function has been called on Stopper.
func (s *Stopper) ShouldStop() chan struct{} {
return s.shouldStopC
}

// Stop signals all managed worker goroutines to stop and wait for them
// to actually stop.
// 停止所有子 goroutine,阻塞等待子 goroutine 退出
func (s *Stopper) Stop() {
s.mu.Lock()
defer s.mu.Unlock()
close(s.shouldStopC)
s.wg.Wait() // 阻塞直到所有子 goroutine 退出
}

// Close closes the internal shouldStopc chan struct{} to signal all
// worker goroutines that they should stop.
// 停止所有子 goroutine,不阻塞
func (s *Stopper) Close() {
close(s.shouldStopC)
}

// Wait waits on the internal sync.WaitGroup. It only return when all
// managed worker goroutines are ready to return and called
// sync.WaitGroup.Done() on the internal sync.WaitGroup.
// 等所有子 goroutine 结束
func (s *Stopper) Wait() {
s.wg.Wait()
}
// stopped 返回 stoper 是否已经被调用 stop 了
func (s *Stopper) stopped() bool {
select {
case <-s.shouldStopC:
return true
default:
}
return false
}