如何优雅地关闭通道

在本文发表数日前，我曾写了一篇文章来解释通道的规则。 那篇文章在reddit和HN上获得了很多点赞，但也有很多人对Go通道的细节设计提出了一些批评意见。

这些批评主要针对于通道设计中的下列细节：

  1. 没有一个简单和通用的方法用来在不改变一个通道的状态的情况下检查这个通道是否已经关闭。
  2. 关闭一个已经关闭的通道将产生一个恐慌，所以在不知道一个通道是否已经关闭的时候关闭此通道是很危险的。
  3. 向一个已关闭的通道发送数据将产生一个恐慌，所以在不知道一个通道是否已经关闭的时候向此通道发送数据是很危险的。

这些批评看上去有几分道理（实际上属于对通道的不正确使用导致的偏见）。 是的，Go语言中并没有提供一个内置函数来检查一个通道是否已经关闭。

在Go中，如果我们能够保证从不会向一个通道发送数据，那么有一个简单的方法来判断此通道是否已经关闭。 此方法已经在上一篇文章通道用例大全中展示过了。 这里为了本文的连贯性，在下面的例子中重新列出了此方法。

  1. package main
  3. import "fmt"
  5. type T int
  7. func IsClosed(ch <-chan T) bool {
  8.     select {
  9.     case <-ch:
  10.         return true
  11.     default:
  12.     }
  14.     return false
  15. }
  17. func main() {
  18.     c := make(chan T)
  19.     fmt.Println(IsClosed(c)) // false
  20.     close(c)
  21.     fmt.Println(IsClosed(c)) // true
  22. }

如前所述，此方法并不是一个通用的检查通道是否已经关闭的方法。

事实上，即使有一个内置closed函数用来检查一个通道是否已经关闭，它的有用性也是十分有限的。 原因是当此函数的一个调用的结果返回时，被查询的通道的状态可能已经又改变了，导致此调用结果并不能反映出被查询的通道的最新状态。 虽然我们可以根据一个调用closed(ch)的返回结果为true而得出我们不应该再向通道ch发送数据的结论， 但是我们不能根据一个调用closed(ch)的返回结果为false而得出我们可以继续向通道ch发送数据的结论。

通道关闭原则

一个常用的使用Go通道的原则是不要在数据接收方或者在有多个发送者的情况下关闭通道。 换句话说，我们只应该让一个通道唯一的发送者关闭此通道。

下面我们将称此原则为通道关闭原则。

当然，这并不是一个通用的关闭通道的原则。通用的原则是不要关闭已关闭的通道。 如果我们能够保证从某个时刻之后，再没有协程将向一个未关闭的非nil通道发送数据，则一个协程可以安全地关闭此通道。 然而，做出这样的保证常常需要很大的努力，从而导致代码过度复杂。 另一方面，遵循通道关闭原则是一件相对简单的事儿。

粗鲁地关闭通道的方法

如果由于某种原因，你一定非要从数据接收方或者让众多发送者中的一个关闭一个通道，你可以使用恢复机制来防止可能产生的恐慌而导致程序崩溃。 下面就是这样的一个实现（假设通道的元素类型为T）。

  1. func SafeClose(ch chan T) (justClosed bool) {
  2.     defer func() {
  3.         if recover() != nil {
  4.             // 一个函数的返回结果可以在defer调用中修改。
  5.             justClosed = false
  6.         }
  7.     }()
  9.     // 假设ch != nil。
  10.     close(ch)   // 如果ch已关闭，则产生一个恐慌。
  11.     return true // <=> justClosed = true; return
  12. }

此方法违反了通道关闭原则。

同样的方法可以用来粗鲁地向一个关闭状态未知的通道发送数据。

  1. func SafeSend(ch chan T, value T) (closed bool) {
  2.     defer func() {
  3.         if recover() != nil {
  4.             closed = true
  5.         }
  6.     }()
  8.     ch <- value  // 如果ch已关闭，则产生一个恐慌。
  9.     return false // <=> closed = false; return
  10. }

这样的粗鲁方法不仅违反了通道关闭原则，而且Go白皮书和标准编译器不保证它的实现中不存在数据竞争。

礼貌地关闭通道的方法

很多Go程序员喜欢使用sync.Once来关闭通道。

  1. type MyChannel struct {
  2.     C    chan T
  3.     once sync.Once
  4. }
  6. func NewMyChannel() *MyChannel {
  7.     return &MyChannel{C: make(chan T)}
  8. }
  10. func (mc *MyChannel) SafeClose() {
  11.     mc.once.Do(func() {
  12.         close(mc.C)
  13.     })
  14. }

当然，我们也可以使用sync.Mutex来防止多次关闭一个通道。

  1. type MyChannel struct {
  2.     C      chan T
  3.     closed bool
  4.     mutex  sync.Mutex
  5. }
  7. func NewMyChannel() *MyChannel {
  8.     return &MyChannel{C: make(chan T)}
  9. }
  11. func (mc *MyChannel) SafeClose() {
  12.     mc.mutex.Lock()
  13.     defer mc.mutex.Unlock()
  14.     if !mc.closed {
  15.         close(mc.C)
  16.         mc.closed = true
  17.     }
  18. }
  20. func (mc *MyChannel) IsClosed() bool {
  21.     mc.mutex.Lock()
  22.     defer mc.mutex.Unlock()
  23.     return mc.closed
  24. }

这些实现确实比上一节中的方法礼貌一些，但是它们不能完全有效地避免数据竞争。 目前的Go白皮书并不保证发生在一个通道上的并发关闭操作和发送操纵不会产生数据竞争。 如果一个SafeClose函数和同一个通道上的发送操作同试运行，则数据竞争可能发生（虽然这样的数据竞争一般并不会带来什么危害）。

优雅地关闭通道的方法

上一节中介绍的SafeSend函数有一个弊端，它的调用不能做为case操作而被使用在select代码块中。 另外，很多Go程序员（包括我）认为上面两节展示的关闭通道的方法不是很优雅。 本节下面将介绍一些在各种情形下使用纯通道操作来关闭通道的方法。

（为了演示程序的完整性，下面这些例子中使用到了sync.WaitGroup。在实践中，sync.WaitGroup并不是必需的。）

情形一：M个接收者和一个发送者。发送者通过关闭用来传输数据的通道来传递发送结束信号

这是最简单的一种情形。当发送者欲结束发送，让它关闭用来传输数据的通道即可。

  1. package main
  3. import (
  4.     "time"
  5.     "math/rand"
  6.     "sync"
  7.     "log"
  8. )
  10. func main() {
  11.     rand.Seed(time.Now().UnixNano())
  12.     log.SetFlags(0)
  14.     // ...
  15.     const Max = 100000
  16.     const NumReceivers = 100
  18.     wgReceivers := sync.WaitGroup{}
  19.     wgReceivers.Add(NumReceivers)
  21.     // ...
  22.     dataCh := make(chan int)
  24.     // 发送者
  25.     go func() {
  26.         for {
  27.             if value := rand.Intn(Max); value == 0 {
  28.                 // 此唯一的发送者可以安全地关闭此数据通道。
  29.                 close(dataCh)
  30.                 return
  31.             } else {
  32.                 dataCh <- value
  33.             }
  34.         }
  35.     }()
  37.     // 接收者
  38.     for i := 0; i < NumReceivers; i++ {
  39.         go func() {
  40.             defer wgReceivers.Done()
  42.             // 接收数据直到通道dataCh已关闭
  43.             // 并且dataCh的缓冲队列已空。
  44.             for value := range dataCh {
  45.                 log.Println(value)
  46.             }
  47.         }()
  48.     }
  50.     wgReceivers.Wait()
  51. }

情形二：一个接收者和N个发送者，此唯一接收者通过关闭一个额外的信号通道来通知发送者不要在发送数据了

此情形比上一种情形复杂一些。我们不能让接收者关闭用来传输数据的通道来停止数据传输，因为这样做违反了通道关闭原则。 但是我们可以让接收者关闭一个额外的信号通道来通知发送者不要在发送数据了。

  1. package main
  3. import (
  4.     "time"
  5.     "math/rand"
  6.     "sync"
  7.     "log"
  8. )
  10. func main() {
  11.     rand.Seed(time.Now().UnixNano())
  12.     log.SetFlags(0)
  14.     // ...
  15.     const Max = 100000
  16.     const NumSenders = 1000
  18.     wgReceivers := sync.WaitGroup{}
  19.     wgReceivers.Add(1)
  21.     // ...
  22.     dataCh := make(chan int)
  23.     stopCh := make(chan struct{})
  24.         // stopCh是一个额外的信号通道。它的
  25.         // 发送者为dataCh数据通道的接收者。
  26.         // 它的接收者为dataCh数据通道的发送者。
  28.     // 发送者
  29.     for i := 0; i < NumSenders; i++ {
  30.         go func() {
  31.             for {
  32.                 // 这里的第一个尝试接收用来让此发送者
  33.                 // 协程尽早地退出。对于这个特定的例子，
  34.                 // 此select代码块并非必需。
  35.                 select {
  36.                 case <- stopCh:
  37.                     return
  38.                 default:
  39.                 }
  41.                 // 即使stopCh已经关闭，此第二个select
  42.                 // 代码块中的第一个分支仍很有可能在若干个
  43.                 // 循环步内依然不会被选中。如果这是不可接受
  44.                 // 的，则上面的第一个select代码块是必需的。
  45.                 select {
  46.                 case <- stopCh:
  47.                     return
  48.                 case dataCh <- rand.Intn(Max):
  49.                 }
  50.             }
  51.         }()
  52.     }
  54.     // 接收者
  55.     go func() {
  56.         defer wgReceivers.Done()
  58.         for value := range dataCh {
  59.             if value == Max-1 {
  60.                 // 此唯一的接收者同时也是stopCh通道的
  61.                 // 唯一发送者。尽管它不能安全地关闭dataCh数
  62.                 // 据通道，但它可以安全地关闭stopCh通道。
  63.                 close(stopCh)
  64.                 return
  65.             }
  67.             log.Println(value)
  68.         }
  69.     }()
  71.     // ...
  72.     wgReceivers.Wait()
  73. }

如此例中的注释所述，对于此额外的信号通道stopCh，它只有一个发送者，即dataCh数据通道的唯一接收者。 dataCh数据通道的接收者关闭了信号通道stopCh，这是不违反通道关闭原则的。

在此例中，数据通道dataCh并没有被关闭。是的，我们不必关闭它。 当一个通道不再被任何协程所使用后，它将逐渐被垃圾回收掉，无论它是否已经被关闭。 所以这里的优雅性体现在通过不关闭一个通道来停止使用此通道。

情形三：M个接收者和N个发送者。它们中的任何协程都可以让一个中间调解协程帮忙发出停止数据传送的信号

这是最复杂的一种情形。我们不能让接收者和发送者中的任何一个关闭用来传输数据的通道，我们也不能让多个接收者之一关闭一个额外的信号通道。 这两种做法都违反了通道关闭原则。 然而，我们可以引入一个中间调解者角色并让其关闭额外的信号通道来通知所有的接收者和发送者结束工作。 具体实现见下例。注意其中使用了一个尝试发送操作来向中间调解者发送信号。

  1. package main
  3. import (
  4.     "time"
  5.     "math/rand"
  6.     "sync"
  7.     "log"
  8.     "strconv"
  9. )
  11. func main() {
  12.     rand.Seed(time.Now().UnixNano())
  13.     log.SetFlags(0)
  15.     // ...
  16.     const Max = 100000
  17.     const NumReceivers = 10
  18.     const NumSenders = 1000
  20.     wgReceivers := sync.WaitGroup{}
  21.     wgReceivers.Add(NumReceivers)
  23.     // ...
  24.     dataCh := make(chan int)
  25.     stopCh := make(chan struct{})
  26.         // stopCh是一个额外的信号通道。它的发送
  27.         // 者为中间调解者。它的接收者为dataCh
  28.         // 数据通道的所有的发送者和接收者。
  29.     toStop := make(chan string, 1)
  30.         // toStop是一个用来通知中间调解者让其
  31.         // 关闭信号通道stopCh的第二个信号通道。
  32.         // 此第二个信号通道的发送者为dataCh数据
  33.         // 通道的所有的发送者和接收者，它的接收者
  34.         // 为中间调解者。它必须为一个缓冲通道。
  36.     var stoppedBy string
  38.     // 中间调解者
  39.     go func() {
  40.         stoppedBy = <-toStop
  41.         close(stopCh)
  42.     }()
  44.     // 发送者
  45.     for i := 0; i < NumSenders; i++ {
  46.         go func(id string) {
  47.             for {
  48.                 value := rand.Intn(Max)
  49.                 if value == 0 {
  50.                     // 为了防止阻塞，这里使用了一个尝试
  51.                     // 发送操作来向中间调解者发送信号。
  52.                     select {
  53.                     case toStop <- "发送者#" + id:
  54.                     default:
  55.                     }
  56.                     return
  57.                 }
  59.                 // 此处的尝试接收操作是为了让此发送协程尽早
  60.                 // 退出。标准编译器对尝试接收和尝试发送做了
  61.                 // 特殊的优化，因而它们的速度很快。
  62.                 select {
  63.                 case <- stopCh:
  64.                     return
  65.                 default:
  66.                 }
  68.                 // 即使stopCh已关闭，如果这个select代码块
  69.                 // 中第二个分支的发送操作是非阻塞的，则第一个
  70.                 // 分支仍很有可能在若干个循环步内依然不会被选
  71.                 // 中。如果这是不可接受的，则上面的第一个尝试
  72.                 // 接收操作代码块是必需的。
  73.                 select {
  74.                 case <- stopCh:
  75.                     return
  76.                 case dataCh <- value:
  77.                 }
  78.             }
  79.         }(strconv.Itoa(i))
  80.     }
  82.     // 接收者
  83.     for i := 0; i < NumReceivers; i++ {
  84.         go func(id string) {
  85.             defer wgReceivers.Done()
  87.             for {
  88.                 // 和发送者协程一样，此处的尝试接收操作是为了
  89.                 // 让此接收协程尽早退出。
  90.                 select {
  91.                 case <- stopCh:
  92.                     return
  93.                 default:
  94.                 }
  96.                 // 即使stopCh已关闭，如果这个select代码块
  97.                 // 中第二个分支的接收操作是非阻塞的，则第一个
  98.                 // 分支仍很有可能在若干个循环步内依然不会被选
  99.                 // 中。如果这是不可接受的，则上面尝试接收操作
  100.                 // 代码块是必需的。
  101.                 select {
  102.                 case <- stopCh:
  103.                     return
  104.                 case value := <-dataCh:
  105.                     if value == Max-1 {
  106.                         // 为了防止阻塞，这里使用了一个尝试
  107.                         // 发送操作来向中间调解者发送信号。
  108.                         select {
  109.                         case toStop <- "接收者#" + id:
  110.                         default:
  111.                         }
  112.                         return
  113.                     }
  115.                     log.Println(value)
  116.                 }
  117.             }
  118.         }(strconv.Itoa(i))
  119.     }
  121.     // ...
  122.     wgReceivers.Wait()
  123.     log.Println("被" + stoppedBy + "终止了")
  124. }

在此例中，通道关闭原则依旧得到了遵守。

请注意，信号通道toStop的容量必须至少为1。 如果它的容量为0，则在中间调解者还未准备好的情况下就已经有某个协程向toStop发送信号时，此信号将被抛弃。

我们也可以不使用尝试发送操作向中间调解者发送信号，但信号通道toStop的容量必须至少为数据发送者和数据接收者的数量之和，以防止向其发送数据时（有一个极其微小的可能）导致某些发送者和接收者协程永久阻塞。

  1. ...
  2. toStop := make(chan string, NumReceivers + NumSenders)
  3. ...
  4.             value := rand.Intn(Max)
  5.             if value == 0 {
  6.                 toStop <- "sender#" + id
  7.                 return
  8.             }
  9. ...
  10.                 if value == Max-1 {
  11.                     toStop <- "receiver#" + id
  12.                     return
  13.                 }
  14. ...

情形四：“M个接收者和一个发送者”情形的一个变种：用来传输数据的通道的关闭请求由第三方发出

有时，数据通道（dataCh）的关闭请求需要由某个第三方协程发出。对于这种情形，我们可以使用一个额外的信号通道来通知唯一的发送者关闭数据通道（dataCh）。

  1. package main
  3. import (
  4.     "time"
  5.     "math/rand"
  6.     "sync"
  7.     "log"
  8. )
  10. func main() {
  11.     rand.Seed(time.Now().UnixNano())
  12.     log.SetFlags(0)
  14.     // ...
  15.     const Max = 100000
  16.     const NumReceivers = 100
  17.     const NumThirdParties = 15
  19.     wgReceivers := sync.WaitGroup{}
  20.     wgReceivers.Add(NumReceivers)
  22.     // ...
  23.     dataCh := make(chan int)
  24.     closing := make(chan struct{}) // 信号通道
  25.     closed := make(chan struct{})
  27.     // 此stop函数可以被安全地多次调用。
  28.     stop := func() {
  29.         select {
  30.         case closing<-struct{}{}:
  31.             <-closed
  32.         case <-closed:
  33.         }
  34.     }
  36.     // 一些第三方协程
  37.     for i := 0; i < NumThirdParties; i++ {
  38.         go func() {
  39.             r := 1 + rand.Intn(3)
  40.             time.Sleep(time.Duration(r) * time.Second)
  41.             stop()
  42.         }()
  43.     }
  45.     // 发送者
  46.     go func() {
  47.         defer func() {
  48.             close(closed)
  49.             close(dataCh)
  50.         }()
  52.         for {
  53.             select{
  54.             case <-closing: return
  55.             default:
  56.             }
  58.             select{
  59.             case <-closing: return
  60.             case dataCh <- rand.Intn(Max):
  61.             }
  62.         }
  63.     }()
  65.     // 接收者
  66.     for i := 0; i < NumReceivers; i++ {
  67.         go func() {
  68.             defer wgReceivers.Done()
  70.             for value := range dataCh {
  71.                 log.Println(value)
  72.             }
  73.         }()
  74.     }
  76.     wgReceivers.Wait()
  77. }

上述代码中的stop函数中使用的技巧偷自Roger Peppe在此贴中的一个留言。

情形五：“N个发送者”的一个变种：用来传输数据的通道必须被关闭以通知各个接收者数据发送已经结束了

在上面的提到的“N个发送者”情形中，为了遵守通道关闭原则，我们避免了关闭数据通道（dataCh）。 但是有时候，数据通道（dataCh）必须被关闭以通知各个接收者数据发送已经结束。 对于这种“N个发送者”情形，我们可以使用一个中间通道将它们转化为“一个发送者”情形，然后继续使用上一节介绍的技巧来关闭此中间通道，从而避免了关闭原始的dataCh数据通道。

  1. package main
  3. import (
  4.     "time"
  5.     "math/rand"
  6.     "sync"
  7.     "log"
  8.     "strconv"
  9. )
  11. func main() {
  12.     rand.Seed(time.Now().UnixNano())
  13.     log.SetFlags(0)
  15.     // ...
  16.     const Max = 1000000
  17.     const NumReceivers = 10
  18.     const NumSenders = 1000
  19.     const NumThirdParties = 15
  21.     wgReceivers := sync.WaitGroup{}
  22.     wgReceivers.Add(NumReceivers)
  24.     // ...
  25.     dataCh := make(chan int)   // 仍就不会被关闭
  26.     middleCh := make(chan int) // 将被关闭
  27.     closing := make(chan string)
  28.     closed := make(chan struct{})
  30.     var stoppedBy string
  32.     stop := func(by string) {
  33.         select {
  34.         case closing <- by:
  35.             <-closed
  36.         case <-closed:
  37.         }
  38.     }
  40.     // 中间层
  41.     go func() {
  42.         exit := func(v int, needSend bool) {
  43.             close(closed)
  44.             if needSend {
  45.                 dataCh <- v
  46.             }
  47.             close(dataCh)
  48.         }
  50.         for {
  51.             select {
  52.             case stoppedBy = <-closing:
  53.                 exit(0, false)
  54.                 return
  55.             case v := <- middleCh:
  56.                 select {
  57.                 case stoppedBy = <-closing:
  58.                     exit(v, true)
  59.                     return
  60.                 case dataCh <- v:
  61.                 }
  62.             }
  63.         }
  64.     }()
  66.     // 一些第三方协程
  67.     for i := 0; i < NumThirdParties; i++ {
  68.         go func(id string) {
  69.             r := 1 + rand.Intn(3)
  70.             time.Sleep(time.Duration(r) * time.Second)
  71.             stop("3rd-party#" + id)
  72.         }(strconv.Itoa(i))
  73.     }
  75.     // 发送者
  76.     for i := 0; i < NumSenders; i++ {
  77.         go func(id string) {
  78.             for {
  79.                 value := rand.Intn(Max)
  80.                 if value == 0 {
  81.                     stop("sender#" + id)
  82.                     return
  83.                 }
  85.                 select {
  86.                 case <- closed:
  87.                     return
  88.                 default:
  89.                 }
  91.                 select {
  92.                 case <- closed:
  93.                     return
  94.                 case middleCh <- value:
  95.                 }
  96.             }
  97.         }(strconv.Itoa(i))
  98.     }
  100.     // 接收者
  101.     for range [NumReceivers]struct{}{} {
  102.         go func() {
  103.             defer wgReceivers.Done()
  105.             for value := range dataCh {
  106.                 log.Println(value)
  107.             }
  108.         }()
  109.     }
  111.     // ...
  112.     wgReceivers.Wait()
  113.     log.Println("stopped by", stoppedBy)
  114. }

更多情形？

在日常编程中可能会遇到更多的变种情形，但是上面介绍的情形是最常见和最基本的。 通过聪明地使用通道（和其它并发同步技术），我相信，对于各种变种，我们总会找到相应的遵守通道关闭原则的解决方法。

结论

并没有什么情况非得逼得我们违反通道关闭原则。 如果你遇到了此情形，请考虑修改你的代码流程和结构设计。

使用通道编程宛如在艺术创作一般！