前言
最近使用go開發(fā)后端服務(wù)，服務(wù)關(guān)閉需要保證channel中的數(shù)據(jù)都被讀取完，理由很簡單，在收到系統(tǒng)的中斷信號后，系統(tǒng)需要做收尾工作，保證channel的數(shù)據(jù)都要被處理掉，然后才可以關(guān)閉系統(tǒng)。但實現(xiàn)起來沒那么簡單，下面來一起看看詳細(xì)的介紹吧。
關(guān)于Go channel設(shè)計和規(guī)范的批評：
- 在不能更改channel狀態(tài)的情況下，沒有簡單普遍的方式來檢查channel是否已經(jīng)關(guān)閉了
- 關(guān)閉已經(jīng)關(guān)閉的channel會導(dǎo)致panic，所以在closer(關(guān)閉者)不知道channel是否已經(jīng)關(guān)閉的情況下去關(guān)閉channel是很危險的
- 發(fā)送值到已經(jīng)關(guān)閉的channel會導(dǎo)致panic，所以如果sender(發(fā)送者)在不知道channel是否已經(jīng)關(guān)閉的情況下去向channel發(fā)送值是很危險的
所以Golang 內(nèi)建的 close 方法可以關(guān)閉 channel，如果往已經(jīng)關(guān)閉的 channel 發(fā)送數(shù)據(jù)，則會報錯：panic: close of closed channel.
看如下代碼，在一段時間內(nèi)，生產(chǎn)者可以不斷往 channel 寫入數(shù)據(jù)，消費者進行處理，一段時間后 channel 關(guān)閉了，這個時候如果還有數(shù)據(jù)往 channel 發(fā)送，程序就會報錯。
package main
import (
"fmt"
"sync"
"time"
)
func main() {
jobs := make(chan int)
var wg sync.WaitGroup
go func() {
time.Sleep(time.Second * 3)
close(jobs)
}()
go func() {
for i := 0; ; i++ {
jobs - i
fmt.Println("produce:", i)
}
}()
wg.Add(1)
go func() {
defer wg.Done()
for i := range jobs {
fmt.Println("consume:", i)
}
}()
wg.Wait()
}
多運行幾次出錯的概率會比較大：
produce: 33334
consume: 33334
consume: 33335
produce: 33335
produce: 33336
consume: 33336
consume: 33337
produce: 33337
produce: 33338
consume: 33338
consume: 33339
produce: 33339
produce: 33340
consume: 33340
panic: send on closed channel
goroutine 19 [running]:
panic(0x49b660, 0xc042410bb0)
C:/Go/src/runtime/panic.go:500 +0x1af
main.main.func2(0xc04203a180)
C:/Users/tanteng/Go/src/examples/channel_close.go:18 +0x6b
created by main.main
C:/Users/tanteng/Go/src/examples/channel_close.go:21 +0xb8
exit status 2
如何優(yōu)雅關(guān)閉 channel
那么在往通道發(fā)數(shù)據(jù)前如何判斷通道是否關(guān)閉呢？
1._,ok := - jobs
此時如果 channel 關(guān)閉，ok 值為 false，如果 channel 沒有關(guān)閉，則會漏掉一個 jobs
2.使用 select 方式
再創(chuàng)建一個 channel，叫做 timeout，如果超時往這個 channel 發(fā)送 true，在生產(chǎn)者發(fā)送數(shù)據(jù)給 jobs 的 channel，用 select 監(jiān)聽 timeout，如果超時則關(guān)閉 jobs 的 channel.
完整代碼如下：
package main
import (
"fmt"
"sync"
"time"
)
func main() {
jobs := make(chan int)
timeout := make(chan bool)
var wg sync.WaitGroup
go func() {
time.Sleep(time.Second * 3)
timeout - true
}()
go func() {
for i := 0; ; i++ {
select {
case -timeout:
close(jobs)
return
default:
jobs - i
fmt.Println("produce:", i)
}
}
}()
wg.Add(1)
go func() {
defer wg.Done()
for i := range jobs {
fmt.Println("consume:", i)
}
}()
wg.Wait()
}
這樣就可以保證不會往已經(jīng)關(guān)閉的 channel 中發(fā)送數(shù)據(jù)了。
總結(jié)
以上就是這篇文章的全部內(nèi)容了，希望本文的內(nèi)容對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價值，如果有疑問大家可以留言交流，謝謝大家對腳本之家的支持。
您可能感興趣的文章:- 基于golang channel實現(xiàn)的輕量級異步任務(wù)分發(fā)器示例代碼
- golang中for循環(huán)遍歷channel時需要注意的問題詳解
- golang實現(xiàn)基于channel的通用連接池詳解
- golang中單向channel的語法介紹
- golang判斷chan channel是否關(guān)閉的方法
- Golang中channel使用的一些小技巧
- Golang中channel的原理解讀(推薦)
