(译) 如何优雅的关闭Go Channel

(译) 如何优雅的关闭Go Channel

Channel关闭原则

不要在消费端关闭channel,不要在有多个并行的生产者时对channel执行关闭操作。

也就是说应该只在[唯一的或者最后唯一剩下]的生产者协程中关闭channel,来通知消费者已经没有值可以继续读了。只要坚持这个原则,就可以确保向一个已经关闭的channel发送数据的情况不可能发生。

暴力关闭channel的正确方法

如果想要在消费端关闭channel,或者在多个生产者端关闭channel,可以使用recover机制来上个保险,避免程序因为panic而崩溃。

1
2
3
4
5
6
7
8
9
10
11
func SafeClose(ch chan T) (justClosed bool) {
defer func() {
if recover() != nil {
justClosed = false
}
}()

// assume ch != nil here.
close(ch) // panic if ch is closed
return true // <=> justClosed = true; return
}

使用这种方法明显违背了上面的channel关闭原则,然后性能还可以,毕竟在每个协程只会调用一次SafeClose,性能损失很小。

同样也可以在生产消息的时候使用recover方法。

1
2
3
4
5
6
7
8
9
10
11
12
func SafeSend(ch chan T, value T) (closed bool) {
defer func() {
if recover() != nil {
// The return result can be altered
// in a defer function call.
closed = true
}
}()

ch <- value // panic if ch is closed
return false // <=> closed = false; return
}

礼貌地关闭channel方法

还有不少人经常使用用sync.Once来关闭channel,这样可以确保只会关闭一次

1
2
3
4
5
6
7
8
9
10
11
12
13
14
type MyChannel struct {
C chan T
once sync.Once
}

func NewMyChannel() *MyChannel {
return &MyChannel{C: make(chan T)}
}

func (mc *MyChannel) SafeClose() {
mc.once.Do(func() {
close(mc.C)
})
}

同样我们也可以使用sync.Mutex达到同样的目的。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
type MyChannel struct {
C chan T
closed bool
mutex sync.Mutex
}

func NewMyChannel() *MyChannel {
return &MyChannel{C: make(chan T)}
}

func (mc *MyChannel) SafeClose() {
mc.mutex.Lock()
if !mc.closed {
close(mc.C)
mc.closed = true
}
mc.mutex.Unlock()
}

func (mc *MyChannel) IsClosed() bool {
mc.mutex.Lock()
defer mc.mutex.Unlock()
return mc.closed
}

要知道golang的设计者不提供SafeClose或者SafeSend方法是有原因的,他们本来就不推荐在消费端或者在并发的多个生产端关闭channel,比如关闭只读channel在语法上就彻底被禁止使用了。

优雅地关闭channel的方法

上文的SafeSend方法一个很大的劣势在于它不能用在select块的case语句中。而另一个很重要的劣势在于像我这样对代码有洁癖的人来说,使用panic/recover和sync/mutex来搞定不是那么的优雅。下面我们引入在不同的场景下可以使用的纯粹的优雅的解决方法。

多个消费者,单个生产者。这种情况最简单,直接让生产者关闭channel好了。

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
package main

import (
"time"
"math/rand"
"sync"
"log"
)

func main() {
rand.Seed(time.Now().UnixNano())
log.SetFlags(0)

// ...
const MaxRandomNumber = 100000
const NumReceivers = 100

wgReceivers := sync.WaitGroup{}
wgReceivers.Add(NumReceivers)

// ...
dataCh := make(chan int, 100)

// the sender
go func() {
for {
if value := rand.Intn(MaxRandomNumber); value == 0 {
// The only sender can close the channel safely.
close(dataCh)
return
} else {
dataCh <- value
}
}
}()

// receivers
for i := 0; i < NumReceivers; i++ {
go func() {
defer wgReceivers.Done()

// Receive values until dataCh is closed and
// the value buffer queue of dataCh is empty.
for value := range dataCh {
log.Println(value)
}
}()
}

wgReceivers.Wait()
}

多个生产者,单个消费者。这种情况要比上面的复杂一点。我们不能在消费端关闭channel,因为这违背了channel关闭原则。但是我们可以让消费端关闭一个附加的信号来通知发送端停止生产数据。

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
package main

import (
"time"
"math/rand"
"sync"
"log"
)

func main() {
rand.Seed(time.Now().UnixNano())
log.SetFlags(0)

// ...
const MaxRandomNumber = 100000
const NumSenders = 1000

wgReceivers := sync.WaitGroup{}
wgReceivers.Add(1)

// ...
dataCh := make(chan int, 100)
stopCh := make(chan struct{})
// stopCh is an additional signal channel.
// Its sender is the receiver of channel dataCh.
// Its reveivers are the senders of channel dataCh.

// senders
for i := 0; i < NumSenders; i++ {
go func() {
for {
// The first select here is to try to exit the goroutine
// as early as possible. In fact, it is not essential
// for this example, so it can be omitted.
select {
case <- stopCh:
return
default:
}

// Even if stopCh is closed, the first branch in the
// second select may be still not selected for some
// loops if the send to dataCh is also unblocked.
// But this is acceptable, so the first select
// can be omitted.
select {
case <- stopCh:
return
case dataCh <- rand.Intn(MaxRandomNumber):
}
}
}()
}

// the receiver
go func() {
defer wgReceivers.Done()

for value := range dataCh {
if value == MaxRandomNumber-1 {
// The receiver of the dataCh channel is
// also the sender of the stopCh cahnnel.
// It is safe to close the stop channel here.
close(stopCh)
return
}

log.Println(value)
}
}()

// ...
wgReceivers.Wait()
}

就上面这个例子,生产者同时也是退出信号channel的接受者,退出信号channel仍然是由它的生产端关闭的,所以这仍然没有违背channel关闭原则。值得注意的是,这个例子中生产端和接受端都没有关闭消息数据的channel,channel在没有任何goroutine引用的时候会自行关闭,而不需要显示进行关闭。

多个生产者,多个消费者

这是最复杂的一种情况,我们既不能让接受端也不能让发送端关闭channel。我们甚至都不能让接受者关闭一个退出信号来通知生产者停止生产。因为我们不能违反channel关闭原则。但是我们可以引入一个额外的协调者来关闭附加的退出信号channel。

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
package main

import (
"time"
"math/rand"
"sync"
"log"
"strconv"
)

func main() {
rand.Seed(time.Now().UnixNano())
log.SetFlags(0)

// ...
const MaxRandomNumber = 100000
const NumReceivers = 10
const NumSenders = 1000

wgReceivers := sync.WaitGroup{}
wgReceivers.Add(NumReceivers)

// ...
dataCh := make(chan int, 100)
stopCh := make(chan struct{})
// stopCh is an additional signal channel.
// Its sender is the moderator goroutine shown below.
// Its reveivers are all senders and receivers of dataCh.
toStop := make(chan string, 1)
// The channel toStop is used to notify the moderator
// to close the additional signal channel (stopCh).
// Its senders are any senders and receivers of dataCh.
// Its reveiver is the moderator goroutine shown below.

var stoppedBy string

// moderator
go func() {
stoppedBy = <- toStop
close(stopCh)
}()

// senders
for i := 0; i < NumSenders; i++ {
go func(id string) {
for {
value := rand.Intn(MaxRandomNumber)
if value == 0 {
// Here, a trick is used to notify the moderator
// to close the additional signal channel.
select {
case toStop <- "sender#" + id:
default:
}
return
}

// The first select here is to try to exit the goroutine
// as early as possible. This select blocks with one
// receive operation case and one default branches will
// be optimized as a try-receive operation by the
// official Go compiler.
select {
case <- stopCh:
return
default:
}

// Even if stopCh is closed, the first branch in the
// second select may be still not selected for some
// loops (and for ever in theory) if the send to
// dataCh is also unblocked.
// This is why the first select block is needed.
select {
case <- stopCh:
return
case dataCh <- value:
}
}
}(strconv.Itoa(i))
}

// receivers
for i := 0; i < NumReceivers; i++ {
go func(id string) {
defer wgReceivers.Done()

for {
// Same as the sender goroutine, the first select here
// is to try to exit the goroutine as early as possible.
select {
case <- stopCh:
return
default:
}

// Even if stopCh is closed, the first branch in the
// second select may be still not selected for some
// loops (and for ever in theory) if the receive from
// dataCh is also unblocked.
// This is why the first select block is needed.
select {
case <- stopCh:
return
case value := <-dataCh:
if value == MaxRandomNumber-1 {
// The same trick is used to notify
// the moderator to close the
// additional signal channel.
select {
case toStop <- "receiver#" + id:
default:
}
return
}

log.Println(value)
}
}
}(strconv.Itoa(i))
}

// ...
wgReceivers.Wait()
log.Println("stopped by", stoppedBy)
}

以上三种场景不能涵盖全部,但是它们是最常见最通用的三种场景,基本上所有的场景都可以划分为以上三类。

总结

首先从三种场景中我们可以得到的结论是永远都是从生产者那边驱动关闭channel,即使不在生产者那边去关闭,也是生产者发送一个关闭的驱动的信息,然后消费者在那边根据回应的消息做出是否要关闭channel,这样才可以channel是被正常关闭。

在多对多的模式中,也是通过生产者驱动channel关闭,但是中间多了一个协调者。为什么要多一个协调者?

因为我们要保证channel仅只被关闭一次,如果是多个生产者的话,谁来做这个关闭操作呢?因此只能是由一个第三方中间人去做这个事情。

情况一:生产者根据条件主动发送事件channel
1
2
3
4
5
6
7
8
9
if value == 0 {
// Here, a trick is used to notify the moderator
// to close the additional signal channel.
select {
case toStop <- "sender#" + id:
default:
}
return
}
情况二:消费者根据生产者的数据条件被动发送事件关闭channel

不过是那边驱动消息发送过来,我认为关闭的条件仍然是由生产者决定,因为多对多的情况下,消费者依然要根据生产者的数据来做出判断,所以决定channel是否关闭,仍然取决于生产者来控制。

1
2
3
4
5
6
7
8
9
10
11
case value := <-dataCh:
if value == MaxRandomNumber-1 {
// The same trick is used to notify
// the moderator to close the
// additional signal channel.
select {
case toStop <- "receiver#" + id:
default:
}
return
}

再来看看channel的关闭原则不要在消费端关闭channel,不要在有多个并行的生产者时对channel执行关闭操作,似乎是印证了我的想法。

评论

`
Your browser is out-of-date!

Update your browser to view this website correctly. Update my browser now

×