go关闭channel的n种方式(关闭时机)总结与分享

短命女 2023-01-18 12:27 675阅读 0赞

channel关闭不当或不关闭会引发很多问题，比如举1个很简单的例子：

两个生成者协程不断给dataCh写数据，main goroutine在持续消费dataCh的数据，这个场景假设每个生产者各生产指定数量数据后需要终止，而如果在第一个和第二个生产者中生产完时都close，或只在其中一个中close dataCh都是不行的，耳熟能详的一个报错：panic: send on closed channel   ； 如果置close于不顾，则又会fatal error: all goroutines are asleep - deadlock!    。。。

且在消费者无法确定每读取完一个数据后是否达到了可关闭的状态时，在消费者协程中关闭dataCh也是不可取的。

关闭channel重点在于关闭的时机、关闭的位置是否恰当。以下基于长期实践和探索，总结出来，各位看官钟爱哪种方式则选用哪种。

先说说最简单的场景：单个ch的发送方是一个goroutine，则发送完毕后直接close即可，代码很简单不展示了；

**目录**

实例场景1：生产的数据数量皆可确定时

策略1：综合各生产者的数量确定总生产数量，根据总值来确定close的时机

策略2：借助所消费的数量在消费方来确定是否已经生产 or  消费完毕

策略3：消费方统计消费总数量是否匹配生产总数量，是则close

实例场景2：通用化处理

策略1：引入信号channel和调度goroutine，协调好生产完毕的时机后关闭数据channel

策略2：增加末尾标记

策略3：增加信号channel通知终止生产协程

--------------------

# 实例场景1：生产的数据数量皆可确定时 #

### 策略1：综合各生产者的数量确定总生产数量，根据总值来确定close的时机 ###

func main() {  // 背景：共生产15枚数据
    	ch := make(chan int)
    	var arr []int
    	go func() {
    		for i := 1; i <= 5; i++ {
    			ch <- i
    			arr = append(arr, i)
    			if len(arr) == 15 {
    				fmt.Println("生产方1:确定我已经生产完毕了，准备关闭...")
    				close(ch)
    			}
    		}
    	}()
    
    	go func() {
    		for i := 6; i <= 10; i++ {
    			ch <- i
    			arr = append(arr, i)
    			if len(arr) == 15 {
    				fmt.Println("生产方2:确定我已经生产完毕了，准备关闭...")
    				close(ch)
    			}
    		}
    	}()
    
    	go func() {
    		for i := 6; i <= 10; i++ {
    			ch <- i
    			arr = append(arr, i)
    			if len(arr) == 15 {
    				fmt.Println("生产方3:确定我已经生产完毕了，准备关闭...")
    				close(ch)
    			}
    		}
    	}()
    
    	for i := range ch {
    		fmt.Println("read:", i)
    	}
    
    	fmt.Println("完成")
    }
    
    
    read: 1
    read: 6
    read: 6
    read: 2
    read: 7
    read: 7
    read: 3
    read: 8
    read: 8
    read: 4
    read: 9
    read: 9
    read: 5
    read: 10
    read: 10
    生产方3:确定我已经生产完毕了，准备关闭...
    完成

清晰明了，应当不难理解；缺点也有，就是生产方需要重复写相同的代码，如果生产方数量较多维护起来也比较麻烦，但一般也多不到哪去，为了保证更强的准确性你也可以给arr加锁；

### 策略2：借助所消费的数量在消费方来确定是否已经生产 or  消费完毕 ###

func main() {  // 随便搞几个生产者
    	ch := make(chan int)
    	var arr []int   
    	go func() {
    		for i := 1; i <= 5; i++ {
    			ch <- i
    		}
    	}()
    
    	go func() {
    		for i := 6; i <= 10; i++ {
    			ch <- i
    		}
    	}()
    
    	for i := range ch {
    		arr = append(arr, i) // map等其它手段也可以，达到目的就好
    		fmt.Println("read:", i)
    		if len(arr) == 10 {
    			fmt.Println("消费方:确定已经消费（生产）完毕了，准备关闭...")
    			close(ch) // 关闭ch，自动解除range阻塞
    		}
    	}
    
    	fmt.Println("完成")
    }
    
    
    read: 6
    read: 7
    read: 1
    read: 8
    read: 2
    read: 9
    read: 3
    read: 10
    read: 4
    read: 5
    消费方:确定已经消费（生产）完毕了，准备关闭...
    完成

该业务背景下我们可以确定生产方共生成10个数据，因此可以依次来精确判断。优势很明显，代码量少、实现起来简单；

再来个稍复杂点的，多个消费者时：

### 策略3：消费方统计消费总数量是否匹配生产总数量，是则close ###

func main() {
    	var mu sync.Mutex  // 确保多个消费者之间统计到的总数精确
    	ch := make(chan int)
    	go func() {
    		for i := 1; i <= 5; i++ {
    			fmt.Println("1-produce: ", i)
    			ch <- i
    		}
    	}()
    
    	go func() {
    		for i := 6; i <= 10; i++ {
    			fmt.Println("2-produce: ", i)
    			ch <- i
    		}
    	}()
    
    	var arr []int
    	go func() {
    	b1:
    		for {
    			select {
    			case i, ok := <-ch:
    				if ok {
    					func() {
    						mu.Lock()
    						defer mu.Unlock()
    						arr = append(arr, i)
    						fmt.Println("1-read:", i)
    					}()
    				} else {
    					break b1
    				}
    			}
    		}
    	}()
    
    	go func() { // 第二个消费者
    	b2:
    		for {
    			select {
    			case i, ok := <-ch:
    				if ok {
    					func() {
    						mu.Lock()
    						defer mu.Unlock()
    						arr = append(arr, i)
    						fmt.Println("2-read:", i)
    					}()
    				} else {
    					break b2
    				}
    			}
    		}
    	}()
    
    	stopCh := make(chan struct{})
    	for {
    		if len(arr) == 10 {
    			close(ch)
    			close(stopCh)
    			fmt.Println("done")
    			break
    		}
    	}
    
    	<-stopCh
    	fmt.Println("完成")
    }

条条大路通罗马。

# 实例场景2：通用化处理 #

### 策略1：引入信号channel和调度goroutine，协调好生产完毕的时机后关闭数据channel ###

以多生产(发送)者多消费(接收)者为例：

func main() {
    	ch := make(chan int)
    	stopCh := make(chan struct{})
    	doneCh := make(chan int)  // 确定生产完毕的信号
    	go func() {
    		for i := 1; i <= 5; i++ {
    			fmt.Println("1-produce: ", i)
    			ch <- i
    		}
    		doneCh <- 1
    	}()
    
    	go func() {
    		for i := 6; i <= 10; i++ {
    			fmt.Println("2-produce: ", i)
    			ch <- i
    		}
    		doneCh <- 2
    	}()
    
    	go func() {
    		m := make(map[int]struct{})
    		for i := range doneCh {
    			m[i] = struct{}{}
    			if len(m) == 2 {  // 是不是全生产完了，来汇总一下就知道了
    				fmt.Println("准备关闭ch：", m)
    				close(ch)  // 生产完毕、关闭数据ch,很稳的
    				break
    			}
    		}
    		fmt.Println("准备关闭stopCh")
    		close(stopCh)
    	}()
    
    	go func() { // 第1个消费者
    	b2:
    		for {
    			select {
    			case i, ok := <-ch:
    				if ok {
    					fmt.Println("1-read:", i)
    				} else {
    					break b2
    				}
    			}
    		}
    	}()
    	go func() { // 第二个消费者
    	b2:
    		for {
    			select {
    			case i, ok := <-ch:
    				if ok {
    					fmt.Println("2-read:", i)
    				} else {
    					break b2
    				}
    			}
    		}
    	}()
    
    	<-stopCh
    	fmt.Println("完成")
    }
    
    
    1-produce:  1
    1-produce:  2
    1-produce:  3
    2-read: 1
    2-read: 3
    1-read: 2
    2-produce:  6
    2-produce:  7
    2-produce:  8
    2-read: 6
    2-read: 8
    1-read: 7
    1-produce:  4
    1-produce:  5
    2-produce:  9
    2-read: 4
    1-read: 5
    2-read: 9
    2-produce:  10
    1-read: 10
    准备关闭ch： map[1:{} 2:{}]
    准备关闭stopCh
    完成

可以继续加固，生产者和消费者区分处理：

func main() {
    	ch := make(chan int)
    	stopCh := make(chan struct{})
    	doneCh1, doneCh2 := make(chan int), make(chan int)
    	go func() {
    		for i := 1; i <= 5; i++ {
    			fmt.Println("1-produce: ", i)
    			ch <- i
    		}
    		doneCh1 <- 1
    	}()
    
    	go func() {
    		for i := 6; i <= 10; i++ {
    			fmt.Println("2-produce: ", i)
    			ch <- i
    		}
    		doneCh1 <- 2
    	}()
    
    	go func() {
    		m := make(map[int]struct{})
    		for i := range doneCh1 {
    			m[i] = struct{}{}
    			if len(m) == 2 {
    				fmt.Println("全部生产完毕,准备关闭数据ch：", m)
    				close(ch) // 很稳的
    				break
    			}
    		}
    	}()
    
    	go func() {
    		m := make(map[int]struct{})
    		for i := range doneCh2 {
    			m[i] = struct{}{}
    			if len(m) == 2 { // 消费者状态全满足后
    				fmt.Println("全部消费完毕,准备关闭stopCh")
    				close(stopCh) // 消费者全部工作完毕，结束程序 很稳的
    				break
    			}
    		}
    	}()
    
    	go func() { // 第1个消费者
    	b2:
    		for {
    			select {
    			case i, ok := <-ch:
    				if ok {
    					fmt.Println("1-read:", i)
    				} else {
    					doneCh2 <- 1
    					break b2
    				}
    			}
    		}
    	}()
    	go func() { // 第二个消费者
    	b2:
    		for {
    			select {
    			case i, ok := <-ch:
    				if ok {
    					fmt.Println("2-read:", i)
    				} else {
    					doneCh2 <- 2
    					break b2
    				}
    			}
    		}
    	}()
    
    	<-stopCh
    	fmt.Println("完成")
    }
    
    1-produce:  1
    1-produce:  2
    1-produce:  3
    2-read: 1
    2-read: 3
    1-read: 2
    2-produce:  6
    2-produce:  7
    2-produce:  8
    2-read: 6
    2-read: 8
    1-read: 7
    1-produce:  4
    1-produce:  5
    2-produce:  9
    2-read: 4
    2-read: 9
    1-read: 5
    2-produce:  10
    2-read: 10
    全部生产完毕,准备关闭数据ch： map[1:{} 2:{}]
    全部消费完毕,准备关闭stopCh
    完成

### 策略2：增加末尾标记 ###

思想与上面类似，写操作完成后增加一个同类型的唯一标记即可：

func main() {
    	ch := make(chan int)
    	go func() {
    		for i := 1; i <= 5; i++ {
    			fmt.Println("produce: ", i)
    			ch <- i
    		}
    		ch <- -1 // 末尾追加唯一值
    	}()
    
    	over:
    	for {
    		select {
    		case i := <-ch:
    			if i == -1 {
    				fmt.Println("准备关闭ch")
    				close(ch)
    				break over
    			}
                fmt.Println("read:", i)
    		}
    	}
    	fmt.Println("完成")
    }
    
    
    produce:  1
    produce:  2
    read: 1
    read: 2
    produce:  3
    produce:  4
    read: 3
    read: 4
    produce:  5
    read: 5
    准备关闭ch
    完成

### 策略3：增加信号channel通知终止生产协程 ###

生产方随时接收不再生产数据的通知，一旦收到通知则使其退出goroutine(当一个通道不再被任何协程所使用后将逐渐被垃圾回收掉，无论是否已经被关闭。)

func main() {
    	ch := make(chan int)
    	stopCh := make(chan struct{})
    	go func() {
    		for i := 1; i <= 5; i++ {
    			select {
    			case <-stopCh:
    				return
    			default:
    
    			}
    
    			select {
    			case <-stopCh:
    				return
    			default:
    				fmt.Println("2-produce: ", i)
    				ch <- i
    
    			}
    		}
    	}()
    
    	go func() {
    		for i := 6; i <= 10; i++ {
    			select {
    			case <-stopCh:
    				return
    			default:
    
    			}
    			select {
    			case <-stopCh:
    				return
    			default:
    				fmt.Println("2-produce: ", i)
    				ch <- i
    			}
    		}
    	}()
    
    over:
    	for {
    		select {
    		case i := <-ch:
    			fmt.Println("read:", i)
    			if i == 8 { // 如需求是读取到8时结束
    				fmt.Println("准备结束")
    				close(stopCh)
    				break over
    			}
    		}
    	}
    	fmt.Println("完成")
    }
    
    
    2-produce:  1
    2-produce:  2
    2-produce:  6
    read: 1
    read: 2
    read: 6
    2-produce:  3
    2-produce:  4
    2-produce:  7
    read: 3
    read: 4
    read: 7
    2-produce:  5
    read: 5
    2-produce:  8
    2-produce:  9
    read: 8
    准备结束
    完成

生产方中之所以有两个读取stopCh的select逻辑是因为select的选取是随机的，即使关闭了stopCh，case <-stopCh也不一定下次就被执行，前面第一个即为了尽早的让goroutine退出。

希望带给你无限帮助哦！

**go并发之美·相关干货好文合集：**

[go并发之美·redis篇·接口幂等性设计实现][go_redis]

[go并发之美·redis篇·实现实时排行榜][go_redis 1]

[go并发日记·多goroutine并发安全的操作共享变量][go_goroutine]

[go并发日记·常用并发场景中的并发控制思路与方案][go]

[go并发日记·避免goroutine泄漏/实现协程可控][go_goroutine 1]

[go并发日记·深入理解go语言channel与goroutine阻塞][go_go_channel_goroutine]

[go并发日记·基于channel优雅的进行并发控制][go_channel]

[go并发日记·WaitGroup并发控制探索及实际场景运用][go_WaitGroup]

[channel死锁的情况分析][channel]

[go并发之美·B操作伴随A操作(B和A同步)][go_B_A_B_A]

[go多函数并发通用工具的设计与实现][go 1]

[go_redis]: https://lan6193.blog.csdn.net/article/details/108938119
[go_redis 1]: https://lan6193.blog.csdn.net/article/details/108931188
[go_goroutine]: https://lan6193.blog.csdn.net/article/details/105715373
[go]: https://lan6193.blog.csdn.net/article/details/108670428
[go_goroutine 1]: https://lan6193.blog.csdn.net/article/details/105559568
[go_go_channel_goroutine]: https://lan6193.blog.csdn.net/article/details/97528721
[go_channel]: https://lan6193.blog.csdn.net/article/details/101201408
[go_WaitGroup]: https://lan6193.blog.csdn.net/article/details/100887891
[channel]: https://lan6193.blog.csdn.net/article/details/97656420
[go_B_A_B_A]: https://lan6193.blog.csdn.net/article/details/108901339
[go 1]: https://lan6193.blog.csdn.net/article/details/103298105