Go channel关闭方法

channel关闭原则

1、不能在消费端关闭channel（基础原则，单生产者或多生产者均不能在消费端关闭）；

2、多个生产者时，不能对channel执行关闭；

3、只有在唯一或最后唯一剩下的生产者协程中关闭channel，之后通知消费者
没有值可以读，才能确保向一个已经关闭的channel中不再发送数据；

暴力关闭channel

强行在消费端或多个生产者端关闭channel会产生pannic，可以使用recover机制接收异常避免崩溃；

生产端：
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 closedreturn false // <=> closed = false; return
}消费端：
func SafeClose(ch chan T) (justClosed bool) {defer func() {if recover() != nil {justClosed = false}}()// assume ch != nil here.close(ch) // panic if ch is closedreturn true // <=> justClosed = true; return
}

不同生产者消费者关闭channel情况

1.单生产者单(多)消费者

直接在生产者端close()；

2.多生产者单消费者

不能在生产者端直接close()，需要新建一个信号channel，通知发送端停止发送数据；channel在没有go协程引用时会自动关闭，不用显式关闭；

package mainimport ("time""math/rand""sync""log"
)func main() {rand.Seed(time.Now().UnixNano())log.SetFlags(0)// ...const MaxRandomNumber = 100000const NumSenders = 1000wgReceivers := 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.// stopCh为添加的信号channel，它的数据来源是dataCh的接受端发出的数据，它的数据    // 是在dataCh的生产端进行消费；// sendersfor 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:returndefault:}// 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.// ？对于某些loop，dataCh的生产端塞入数据，即使stopCh已经关闭，第二个select的           // 第一个分支仍然可能不能被选择(select如果多个条件同时满足条件，会随机选择)；select {case <- stopCh:returncase dataCh <- rand.Intn(MaxRandomNumber):}}}()}// the receivergo 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()
}

3.多生产者多消费者

不能让接受端或发送端关闭channel，甚至都不能让接受者关闭一个退出信号来通知生产者停止生产，因为多消费者会导致接受者多次执行close()，相当于多个生产端关闭channel，违反了channel关闭原则，但可以引入一个额外的协调者来关闭附加的退出信号channel。

package mainimport ("time""math/rand""sync""log""strconv"
)func main() {rand.Seed(time.Now().UnixNano())log.SetFlags(0)// ...const MaxRandomNumber = 100000const NumReceivers = 10const NumSenders = 1000wgReceivers := 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// moderatorgo func() {stoppedBy = <- toStopclose(stopCh)}()// sendersfor 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:returndefault:}// 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:returncase dataCh <- value:}}}(strconv.Itoa(i))}// receiversfor 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:returndefault:}// 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:returncase 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)
}

Context结束多个协程

package mainimport ("context""fmt""sync""time"
)func worker(ctx context.Context, id int, wg *sync.WaitGroup) {defer wg.Done()for {select {case <-ctx.Done():fmt.Printf("Worker %d canceled\n", id)returndefault:// 执行协程的工作任务fmt.Printf("Worker %d working\n", id)time.Sleep(time.Second)}}
}func main() {ctx, cancel := context.WithCancel(context.Background())var wg sync.WaitGroup// 启动多个协程for i := 1; i <= 5; i++ {wg.Add(1)go worker(ctx, i, &wg)}// 主程序等待一段时间后取消所有协程time.Sleep(time.Second * 3)cancel()// 等待所有协程完成wg.Wait()fmt.Println("Main program finished")
}