在本文发表数日前,我曾写了一篇文章来解释通道的规则。 那篇文章在reddit和HN上获得了很多点赞,但也有很多人对Go通道的细节设计提出了一些批评意见。
这些批评主要针对于通道设计中的下列细节:
这些批评看上去有几分道理(实际上属于对通道的不正确使用导致的偏见)。 是的,Go语言中并没有提供一个内置函数来检查一个通道是否已经关闭。
package main
import "fmt"
type T int
func IsClosed(ch <-chan T) bool {
select {
case <-ch:
return true
default:
}
return false
}
func main() {
c := make(chan T)
fmt.Println(IsClosed(c)) // false
close(c)
fmt.Println(IsClosed(c)) // true
}
如前所述,此方法并不是一个通用的检查通道是否已经关闭的方法。
事实上,即使有一个内置closed
函数用来检查一个通道是否已经关闭,它的有用性也是十分有限的。
原因是当此函数的一个调用的结果返回时,被查询的通道的状态可能已经又改变了,导致此调用结果并不能反映出被查询的通道的最新状态。
虽然我们可以根据一个调用closed(ch)
的返回结果为true
而得出我们不应该再向通道ch
发送数据的结论,
但是我们不能根据一个调用closed(ch)
的返回结果为false
而得出我们可以继续向通道ch
发送数据的结论。
一个常用的使用Go通道的原则是不要在数据接收方或者在有多个发送者的情况下关闭通道。 换句话说,我们只应该让一个通道唯一的发送者关闭此通道。
下面我们将称此原则为通道关闭原则。
当然,这并不是一个通用的关闭通道的原则。通用的原则是不要关闭已关闭的通道。 如果我们能够保证从某个时刻之后,再没有协程将向一个未关闭的非nil通道发送数据,则一个协程可以安全地关闭此通道。 然而,做出这样的保证常常需要很大的努力,从而导致代码过度复杂。 另一方面,遵循通道关闭原则是一件相对简单的事儿。
T
)。
func SafeClose(ch chan T) (justClosed bool) {
defer func() {
if recover() != nil {
// 一个函数的返回结果可以在defer调用中修改。
justClosed = false
}
}()
// 假设ch != nil。
close(ch) // 如果ch已关闭,则产生一个恐慌。
return true // <=> justClosed = true; return
}
此方法违反了通道关闭原则。
同样的方法可以用来粗鲁地向一个关闭状态未知的通道发送数据。func SafeSend(ch chan T, value T) (closed bool) {
defer func() {
if recover() != nil {
closed = true
}
}()
ch <- value // 如果ch已关闭,则产生一个恐慌。
return false // <=> closed = false; return
}
sync.Once
来关闭通道。
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
来防止多次关闭一个通道。
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()
defer mc.mutex.Unlock()
if !mc.closed {
close(mc.C)
mc.closed = true
}
}
func (mc *MyChannel) IsClosed() bool {
mc.mutex.Lock()
defer mc.mutex.Unlock()
return mc.closed
}
这些实现确实比上一节中的方法礼貌一些,但是它们不能完全有效地避免数据竞争。
目前的Go白皮书并不保证发生在一个通道上的并发关闭操作和发送操作不会产生数据竞争。
如果一个SafeClose
函数和同一个通道上的发送操作同时运行,则数据竞争可能发生(虽然这样的数据竞争一般并不会带来什么危害)。
上一节中介绍的SafeSend
函数有一个弊端,它的调用不能做为case
操作而被使用在select
代码块中。
另外,很多Go程序员(包括我)认为上面两节展示的关闭通道的方法不是很优雅。
本节下面将介绍一些在各种情形下使用纯通道操作来关闭通道的方法。
(为了演示程序的完整性,下面这些例子中使用到了sync.WaitGroup
。在实践中,sync.WaitGroup
并不是必需的。)
package main
import (
"time"
"math/rand"
"sync"
"log"
)
func main() {
rand.Seed(time.Now().UnixNano()) // Go 1.20之前需要
log.SetFlags(0)
// ...
const Max = 100000
const NumReceivers = 100
wgReceivers := sync.WaitGroup{}
wgReceivers.Add(NumReceivers)
// ...
dataCh := make(chan int)
// 发送者
go func() {
for {
if value := rand.Intn(Max); value == 0 {
// 此唯一的发送者可以安全地关闭此数据通道。
close(dataCh)
return
} else {
dataCh <- value
}
}
}()
// 接收者
for i := 0; i < NumReceivers; i++ {
go func() {
defer wgReceivers.Done()
// 接收数据直到通道dataCh已关闭
// 并且dataCh的缓冲队列已空。
for value := range dataCh {
log.Println(value)
}
}()
}
wgReceivers.Wait()
}
package main
import (
"time"
"math/rand"
"sync"
"log"
)
func main() {
rand.Seed(time.Now().UnixNano()) // Go 1.20之前需要
log.SetFlags(0)
// ...
const Max = 100000
const NumSenders = 1000
wgReceivers := sync.WaitGroup{}
wgReceivers.Add(1)
// ...
dataCh := make(chan int)
stopCh := make(chan struct{})
// stopCh是一个额外的信号通道。它的
// 发送者为dataCh数据通道的接收者。
// 它的接收者为dataCh数据通道的发送者。
// 发送者
for i := 0; i < NumSenders; i++ {
go func() {
for {
// 这里的第一个尝试接收用来让此发送者
// 协程尽早地退出。对于这个特定的例子,
// 此select代码块并非必需。
select {
case <- stopCh:
return
default:
}
// 即使stopCh已经关闭,此第二个select
// 代码块中的第一个分支仍很有可能在若干个
// 循环步内依然不会被选中。如果这是不可接受
// 的,则上面的第一个select代码块是必需的。
select {
case <- stopCh:
return
case dataCh <- rand.Intn(Max):
}
}
}()
}
// 接收者
go func() {
defer wgReceivers.Done()
for value := range dataCh {
if value == Max-1 {
// 此唯一的接收者同时也是stopCh通道的
// 唯一发送者。尽管它不能安全地关闭dataCh数
// 据通道,但它可以安全地关闭stopCh通道。
close(stopCh)
return
}
log.Println(value)
}
}()
// ...
wgReceivers.Wait()
}
如此例中的注释所述,对于此额外的信号通道stopCh
,它只有一个发送者,即dataCh
数据通道的唯一接收者。
dataCh
数据通道的接收者关闭了信号通道stopCh
,这是不违反通道关闭原则的。
在此例中,数据通道dataCh
并没有被关闭。是的,我们不必关闭它。
当一个通道不再被任何协程所使用后,它将逐渐被垃圾回收掉,无论它是否已经被关闭。
所以这里的优雅性体现在通过不关闭一个通道来停止使用此通道。
package main
import (
"time"
"math/rand"
"sync"
"log"
"strconv"
)
func main() {
rand.Seed(time.Now().UnixNano()) // Go 1.20之前需要
log.SetFlags(0)
// ...
const Max = 100000
const NumReceivers = 10
const NumSenders = 1000
wgReceivers := sync.WaitGroup{}
wgReceivers.Add(NumReceivers)
// ...
dataCh := make(chan int)
stopCh := make(chan struct{})
// stopCh是一个额外的信号通道。它的发送
// 者为中间调解者。它的接收者为dataCh
// 数据通道的所有的发送者和接收者。
toStop := make(chan string, 1)
// toStop是一个用来通知中间调解者让其
// 关闭信号通道stopCh的第二个信号通道。
// 此第二个信号通道的发送者为dataCh数据
// 通道的所有的发送者和接收者,它的接收者
// 为中间调解者。它必须为一个缓冲通道。
var stoppedBy string
// 中间调解者
go func() {
stoppedBy = <-toStop
close(stopCh)
}()
// 发送者
for i := 0; i < NumSenders; i++ {
go func(id string) {
for {
value := rand.Intn(Max)
if value == 0 {
// 为了防止阻塞,这里使用了一个尝试
// 发送操作来向中间调解者发送信号。
select {
case toStop <- "发送者#" + id:
default:
}
return
}
// 此处的尝试接收操作是为了让此发送协程尽早
// 退出。标准编译器对尝试接收和尝试发送做了
// 特殊的优化,因而它们的速度很快。
select {
case <- stopCh:
return
default:
}
// 即使stopCh已关闭,如果这个select代码块
// 中第二个分支的发送操作是非阻塞的,则第一个
// 分支仍很有可能在若干个循环步内依然不会被选
// 中。如果这是不可接受的,则上面的第一个尝试
// 接收操作代码块是必需的。
select {
case <- stopCh:
return
case dataCh <- value:
}
}
}(strconv.Itoa(i))
}
// 接收者
for i := 0; i < NumReceivers; i++ {
go func(id string) {
defer wgReceivers.Done()
for {
// 和发送者协程一样,此处的尝试接收操作是为了
// 让此接收协程尽早退出。
select {
case <- stopCh:
return
default:
}
// 即使stopCh已关闭,如果这个select代码块
// 中第二个分支的接收操作是非阻塞的,则第一个
// 分支仍很有可能在若干个循环步内依然不会被选
// 中。如果这是不可接受的,则上面尝试接收操作
// 代码块是必需的。
select {
case <- stopCh:
return
case value := <-dataCh:
if value == Max-1 {
// 为了防止阻塞,这里使用了一个尝试
// 发送操作来向中间调解者发送信号。
select {
case toStop <- "接收者#" + id:
default:
}
return
}
log.Println(value)
}
}
}(strconv.Itoa(i))
}
// ...
wgReceivers.Wait()
log.Println("被" + stoppedBy + "终止了")
}
在此例中,通道关闭原则依旧得到了遵守。
请注意,信号通道toStop
的容量必须至少为1。
如果它的容量为0,则在中间调解者还未准备好的情况下就已经有某个协程向toStop
发送信号时,此信号将被抛弃。
toStop
的容量必须至少为数据发送者和数据接收者的数量之和,以防止向其发送数据时(有一个极其微小的可能)导致某些发送者和接收者协程永久阻塞。
...
toStop := make(chan string, NumReceivers + NumSenders)
...
value := rand.Intn(Max)
if value == 0 {
toStop <- "sender#" + id
return
}
...
if value == Max-1 {
toStop <- "receiver#" + id
return
}
...
dataCh
)的关闭请求需要由某个第三方协程发出。对于这种情形,我们可以使用一个额外的信号通道来通知唯一的发送者关闭数据通道(dataCh
)。
package main
import (
"time"
"math/rand"
"sync"
"log"
)
func main() {
rand.Seed(time.Now().UnixNano()) // Go 1.20之前需要
log.SetFlags(0)
// ...
const Max = 100000
const NumReceivers = 100
const NumThirdParties = 15
wgReceivers := sync.WaitGroup{}
wgReceivers.Add(NumReceivers)
// ...
dataCh := make(chan int)
closing := make(chan struct{}) // 信号通道
closed := make(chan struct{})
// 此stop函数可以被安全地多次调用。
stop := func() {
select {
case closing<-struct{}{}:
<-closed
case <-closed:
}
}
// 一些第三方协程
for i := 0; i < NumThirdParties; i++ {
go func() {
r := 1 + rand.Intn(3)
time.Sleep(time.Duration(r) * time.Second)
stop()
}()
}
// 发送者
go func() {
defer func() {
close(closed)
close(dataCh)
}()
for {
select{
case <-closing: return
default:
}
select{
case <-closing: return
case dataCh <- rand.Intn(Max):
}
}
}()
// 接收者
for i := 0; i < NumReceivers; i++ {
go func() {
defer wgReceivers.Done()
for value := range dataCh {
log.Println(value)
}
}()
}
wgReceivers.Wait()
}
上述代码中的stop
函数中使用的技巧偷自Roger Peppe在此贴中的一个留言。
dataCh
)。
但是有时候,数据通道(dataCh
)必须被关闭以通知各个接收者数据发送已经结束。
对于这种“N个发送者”情形,我们可以使用一个中间通道将它们转化为“一个发送者”情形,然后继续使用上一节介绍的技巧来关闭此中间通道,从而避免了关闭原始的dataCh
数据通道。
package main
import (
"time"
"math/rand"
"sync"
"log"
"strconv"
)
func main() {
rand.Seed(time.Now().UnixNano()) // Go 1.20之前需要
log.SetFlags(0)
// ...
const Max = 1000000
const NumReceivers = 10
const NumSenders = 1000
const NumThirdParties = 15
wgReceivers := sync.WaitGroup{}
wgReceivers.Add(NumReceivers)
// ...
dataCh := make(chan int) // 将被关闭
middleCh := make(chan int) // 不会被关闭
closing := make(chan string)
closed := make(chan struct{})
var stoppedBy string
stop := func(by string) {
select {
case closing <- by:
<-closed
case <-closed:
}
}
// 中间层
go func() {
exit := func(v int, needSend bool) {
close(closed)
if needSend {
dataCh <- v
}
close(dataCh)
}
for {
select {
case stoppedBy = <-closing:
exit(0, false)
return
case v := <- middleCh:
select {
case stoppedBy = <-closing:
exit(v, true)
return
case dataCh <- v:
}
}
}
}()
// 一些第三方协程
for i := 0; i < NumThirdParties; i++ {
go func(id string) {
r := 1 + rand.Intn(3)
time.Sleep(time.Duration(r) * time.Second)
stop("3rd-party#" + id)
}(strconv.Itoa(i))
}
// 发送者
for i := 0; i < NumSenders; i++ {
go func(id string) {
for {
value := rand.Intn(Max)
if value == 0 {
stop("sender#" + id)
return
}
select {
case <- closed:
return
default:
}
select {
case <- closed:
return
case middleCh <- value:
}
}
}(strconv.Itoa(i))
}
// 接收者
for range [NumReceivers]struct{}{} {
go func() {
defer wgReceivers.Done()
for value := range dataCh {
log.Println(value)
}
}()
}
// ...
wgReceivers.Wait()
log.Println("stopped by", stoppedBy)
}
在日常编程中可能会遇到更多的变种情形,但是上面介绍的情形是最常见和最基本的。 通过聪明地使用通道(和其它并发同步技术),我相信,对于各种变种,我们总会找到相应的遵守通道关闭原则的解决方法。
并没有什么情况非得逼得我们违反通道关闭原则。 如果你遇到了此情形,请考虑修改你的代码流程和结构设计。
使用通道编程宛如在艺术创作一般!
Go101.org网站内容包括Go编程各种相关知识(比如Go基础、Go优化、Go细节、Go实战、Go测验、Go工具等)。后续将不断有新的内容加入。敬请收藏关注期待。
本丛书微信公众号(联系方式一)名称为"Go 101"。二维码在网站首页。此公众号将时不时地发表一些Go语言相关的原创短文。各位如果感兴趣,可以搜索关注一下。
《Go语言101》系列丛书项目目前托管在Github上(联系方式二)。欢迎各位在此项目中通过提交bug和PR的方式来改进完善《Go语言101》丛书中的各篇文章。我们可以在项目目录下运行go run .
来浏览和确认各种改动。
本书的twitter帐号为@Golang_101(联系方式三)。玩推的Go友可以适当关注。
你或许对本书作者老貘开发的一些App感兴趣。
sync
标准库包sync/atomic
标准库包