channel的关闭的经典场景
关闭原则
一个常用的使用Go通道的原则是不要在数据接收方或者在有多个发送者的情况下关闭通
道。
通用的原则是不要关闭已关闭的通道
SRE实战 互联网时代守护先锋，助力企业售后服务体系运筹帷幄！一键直达领取阿里云限量特价优惠。错误关闭
ch:=make(chan int,1)
go func(ch chan int) {
ch<-1
log.Println("OK1")
ch<-2 //阻塞1秒
log.Println("OK2")
ch<-3 //报错 因为下面把管道关闭了
}(ch)
go func(ch chan int) {
log.Println("进入读")
time.Sleep(time.Second)
log.Println(<-ch)
close(ch)
}(ch)
time.Sleep(time.Hour)
优雅关闭
do once代表只能执行一次，防止重复关闭
type MyChannel struct {
C chan int
once sync.Once
}
func (myChannel MyChannel) SafeClose(){
myChannel.once.Do(func() {
close(myChannel.C)
}) //once代表只执行一次
}
func main() {
write:=MyChannel{C:make(chan int)}
go func() {write.C<-1;write.SafeClose()}()
go func() {log.Println(<-write.C)}()
time.Sleep(time.Hour)
}
关闭情况
1、多个接收者，一个发送者
这里只有一个发送者，当value=0时，这个发送者关闭管道，这里只有一个写进程，所以不用担心重复关闭问题，写进程可以直接关闭管道
这时，所有的接受者for-range会全部结束
这里的结束信号由写者发出
func Write(dataCh chan int){
for{
if value:=rand.Intn(1000);value!=0{
time.Sleep(time.Microsecond*500)
dataCh<-value
}else{
close(dataCh)
}
}
}
const NumReceivers =100
var wg=sync.WaitGroup{}
//多个接收者
func Readers(dataCh <-chan int){
for i:=0;i<NumReceivers;i++{
go func() {
defer wg.Done()
for value:=range dataCh{
log.Println(value)
}
}()
}
}
func main() {
rand.Seed(time.Now().UnixNano())
wg.Add(NumReceivers)
dataCh:=make(chan int)
go Write(dataCh)
Readers(dataCh)
wg.Wait()
}
2、【变种】多个接收者，一个发送者，但是关闭信号是由第三方发送
这里用了两个关闭信号
closing：用来通知写携程可以停止写了
closed：用来表示closing信号已经由一个第三方发出过了，后面的第三方可以不用再发了
package main
import (
"log"
"math/rand"
"sync"
"time"
)
func main() {
rand.Seed(time.Now().UnixNano())
const NumReceivers = 100
const NumThirdParties = 100 //第三方
wg4:=sync.WaitGroup{}
wg4.Add(NumReceivers)
dataCh2:=make(chan int)
closing:=make(chan bool) //通知写进程 关闭的信号
closed:=make(chan bool) //通道已经关闭的 信号
stop:= func() {
select {
case closing<-true: //通知关闭
<-closed //有一个写进程收到通知 则结束
case <-closed: //如果已经关闭 则执行这一句 所以多次调用stop是安全非阻塞的
}
}
//多个第三方携程
for i:=0;i<NumThirdParties;i++{
go func() {
time.Sleep(time.Second*time.Duration(rand.Intn(100)))
stop()
}()
}
//发送者 1
go func() {
defer func() {
close(closed) //通知第三方 已经关闭了
close(dataCh2) //关闭数据通道
}()
for{
select {
case <-closing:
return
default:
}
select {
case <-closing: //收到第三方的关闭请求
return
case dataCh2<-rand.Intn(100000):
}
}
}()
for i:=0;i<NumReceivers;i++{
go func() {
defer wg4.Done()
for value:=range dataCh2{
log.Println(value)
}
}()
}
wg4.Wait()
}
3、多个发送者，一个接收者
这里有10000个发送者，为了遵循 管道不能被读者关闭的原则，于是这里又建立了一个stopCh非缓冲管道，目的只是为了通知多个发送者，可以不用发送了，发送者先检查有没有关闭信号，然后再发送。这里用了两次select，来检查 。
同时注意，为了防止重复关闭管道，这里没有让写进程关闭dataCh，当所有读写进程都不再引用dataCh的时候，就会被gc
这里的关闭信号由读者发出
const NumSenders = 10000
var wg2 = sync.WaitGroup{}
var dataCh = make(chan int)
var stopCh = make(chan bool)
func Sender(){
for i:=0;i<NumSenders;i++{
go func() {
select {
case <-stopCh: //通道关闭后直接执行这个
return
default:
}
select {
case <-stopCh: //当通道被关闭 则直接执行这个
return //直接返回
case dataCh<-rand.Intn(1000):
}
}()
}
}
func Receive(){
defer wg2.Done()
for value:=range dataCh{
log.Println(value)
if value==0{
close(stopCh)
return
}
}
}
func main() {
wg2.Add(1)
rand.Seed(time.Now().UnixNano())
go Sender()
go Receive()
wg2.Wait()
}
4、多个接收者和多个发送者
这种情况比较复杂，必须叫一个中间人来通知双方，当读写进程需要关闭，通过toStop管道通知中间人，中间人收到通知然后关闭stopCh2管道，这个管道是让读写进程结束读写的信号。
toStop的容量必须设置成**>=1** 如果设置成非缓冲管道，那么如果中间人没有准备好，要发出结束信号时会阻塞，就会转到default,从而让结束信号丢失。
这里的结束信号读写进程都可以发出
var NumSends = 100
var NumRecives = 50
var wg3 = sync.WaitGroup{}
func main() {
rand.Seed(time.Now().UnixNano())
wg3.Add(NumRecives)
dataCh2 := make(chan int)
stopCh2 := make(chan bool)
toStop := make(chan bool, 1) //通知中间者可以关闭了 至少为1
//中间人
go func() {
_ = <-toStop
close(stopCh2)
}()
//发送者
for i := 0; i < NumSends; i++ {
go func() {
for {
v := rand.Intn(1000)
if v == 0 { //写进程通知关闭
select {
case toStop <- true:
dataCh2 <- v
log.Println("===========================", 0, "-")
default:
}
return
}
//尝试关闭
select {
case <-stopCh2:
return
default:
}
select {
case <-stopCh2:
return
case dataCh2 <- v:
log.Println(v)
}
}
}()
}
//接收者
for i := 0; i < NumRecives; i++ {
go func() {
defer wg3.Done()
for {
select {
case <-stopCh2:
return
default:
}
select {
case <-stopCh2:
return
case val := <-dataCh2:
log.Println(val)
if val == 888 { //读进程通知关闭
select {
case toStop <- true:
log.Println("==========================", 888, "-")
default:
}
return
}
}
}
}()
}
wg3.Wait()
}
更多精彩
