概要

在 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
}