本文介紹了使用 golang channel 的諸多特性和技巧，已經熟悉了 go 語言特性的小伙伴也可以看看，很有啟發(fā)。 不同于傳統(tǒng)的多線程并發(fā)模型使用共享內存來實現線程間通信的方式，golang 的哲學是通過 channel 進行協(xié)程 (goroutine) 之間的通信來實現數據共享：

Do not communicate by sharing memory; instead, share memory by communicating.

這種方式的優(yōu)點是通過提供原子的通信原語，避免了競態(tài)情形 (race condition) 下復雜的鎖機制。channel 可以看成一個 FIFO 隊列，對 FIFO 隊列的讀寫都是原子的操作，不需要加鎖。對 channel 的操作行為結果總結如下：

讀取一個已關閉的 channel 時，總是能讀取到對應類型的零值，為了和讀取非空未關閉 channel 的行為區(qū)別，可以使用兩個接收值：

//okisfalsewhenchisclosed v,ok:=<-chgolang 中大部分類型都是值類型（只有 slice / channel / map 是引用類型），讀/寫類型是值類型的 channel 時，如果元素 size 比較大時，應該使用指針代替，避免頻繁的內存拷貝開銷。

內部實現

如圖所示，在 channel 的內部實現中（具體定義在$GOROOT/src/runtime/chan.go里），維護了 3 個隊列：

讀等待協(xié)程隊列 recvq，維護了阻塞在讀此 channel 的協(xié)程列表

寫等待協(xié)程隊列 sendq，維護了阻塞在寫此 channel 的協(xié)程列表

緩沖數據隊列 buf，用環(huán)形隊列實現，不帶緩沖的 channel 此隊列 size 則為 0

img 當協(xié)程嘗試從未關閉的 channel 中讀取數據時，內部的操作如下：

當 buf 非空時，此時 recvq 必為空，buf 彈出一個元素給讀協(xié)程，讀協(xié)程獲得數據后繼續(xù)執(zhí)行，此時若 sendq 非空，則從 sendq 中彈出一個寫協(xié)程轉入 running 狀態(tài)，待寫數據入隊列 buf ，此時讀取操作<- ch?未阻塞；

當 buf 為空但 sendq 非空時（不帶緩沖的 channel)，則從 sendq 中彈出一個寫協(xié)程轉入 running 狀態(tài)，待寫數據直接傳遞給讀協(xié)程，讀協(xié)程繼續(xù)執(zhí)行，此時讀取操作<- ch?未阻塞；

當 buf 為空并且 sendq 也為空時，讀協(xié)程入隊列 recvq 并轉入 blocking 狀態(tài)，當后續(xù)有其他協(xié)程往 channel 寫數據時，讀協(xié)程才會重新轉入 running 狀態(tài)，此時讀取操作<- ch?阻塞。

類似的，當協(xié)程嘗試往未關閉的 channel 中寫入數據時，內部的操作如下：

當隊列 recvq 非空時，此時隊列 buf 必為空，從 recvq 彈出一個讀協(xié)程接收待寫數據，此讀協(xié)程此時結束阻塞并轉入 running 狀態(tài)，寫協(xié)程繼續(xù)執(zhí)行，此時寫入操作ch <-?未阻塞；

當隊列 recvq 為空但 buf 未滿時，此時 sendq 必為空，寫協(xié)程的待寫數據入 buf 然后繼續(xù)執(zhí)行，此時寫入操作ch <-?未阻塞；

當隊列 recvq 為空并且 buf 為滿時，此時寫協(xié)程入隊列 sendq 并轉入 blokcing 狀態(tài)，當后續(xù)有其他協(xié)程從 channel 中讀數據時，寫協(xié)程才會重新轉入 running 狀態(tài)，此時寫入操作ch <-?阻塞。

當關閉 non-nil channel 時，內部的操作如下：

當隊列 recvq 非空時，此時 buf 必為空，recvq 中的所有協(xié)程都將收到對應類型的零值然后結束阻塞狀態(tài)；

當隊列 sendq 非空時，此時 buf 必為滿，sendq 中的所有協(xié)程都會產生 panic ，在 buf 中數據仍然會保留直到被其他協(xié)程讀取。

使用場景

除了常規(guī)的用來在協(xié)程之間傳遞數據外，本節(jié)列出了一些特殊的使用 channel 的場景。

futures / promises

golang 雖然沒有直接提供 futrue / promise 模型的操作原語，但通過 goroutine 和 channel 可以實現類似的功能：

packagemain import( "io/ioutil" "log" "net/http" ) //RequestFuture,httprequestpromise. funcRequestFuture(urlstring)<-chan?[]byte?{ ????c?:=?make(chan?[]byte,?1) ????go?func()?{ ????????var?body?[]byte ????????defer?func()?{ ????????????c?<-?body ????????}() ????????res,?err?:=?http.Get(url) ????????if?err?!=?nil?{ ????????????return ????????} ????????defer?res.Body.Close() ????????body,?_?=?ioutil.ReadAll(res.Body) ????}() ????return?c } func?main()?{ ????future?:=?RequestFuture("https://api.github.com/users/octocat/orgs") ????body?:=?<-future ????log.Printf("reponse?length:?%d",?len(body)) }

條件變量 (condition variable)

類型于 POSIX 接口中線程通知其他線程某個事件發(fā)生的條件變量，channel 的特性也可以用來當成協(xié)程之間同步的條件變量。因為 channel 只是用來通知，所以 channel 中具體的數據類型和值并不重要，這種場景一般用strct {}作為 channel 的類型。

一對一通知

類似pthread_cond_signal()的功能，用來在一個協(xié)程中通知另個某一個協(xié)程事件發(fā)生：

packagemain import( "fmt" "time" ) funcmain(){ ch:=make(chanstruct{}) nums:=make([]int,100) gofunc(){ time.Sleep(time.Second) fori:=0;i

廣播通知

類似pthread_cond_broadcast()的功能。利用從已關閉的 channel 讀取數據時總是非阻塞的特性，可以實現在一個協(xié)程中向其他多個協(xié)程廣播某個事件發(fā)生的通知：

packagemain import( "fmt" "time" ) funcmain(){ N:=10 exit:=make(chanstruct{}) done:=make(chanstruct{},N) //startNworkergoroutines fori:=0;i

信號量

channel 的讀/寫相當于信號量的 P / V 操作，下面的示例程序中 channel 相當于信號量：

packagemain import( "log" "math/rand" "time" ) typeSeatint typeBarchanSeat func(barBar)ServeConsumer(customerIdint){ log.Print("->consumer#",customerId,"entersthebar") seat:=<-bar?//?need?a?seat?to?drink ????log.Print("consumer#",?customerId,?"?drinks?at?seat#",?seat) ????time.Sleep(time.Second?*?time.Duration(2+rand.Intn(6))) ????log.Print("<-?consumer#",?customerId,?"?frees?seat#",?seat) ????bar?<-?seat?//?free?the?seat?and?leave?the?bar } func?main()?{ ????rand.Seed(time.Now().UnixNano()) ????bar24x7?:=?make(Bar,?10)?//?the?bar?has?10?seats ????//?Place?seats?in?an?bar. ????for?seatId?:=?0;?seatId?

互斥量

互斥量相當于二元信號里，所以 cap 為 1 的 channel 可以當成互斥量使用：

packagemain import"fmt" funcmain(){ mutex:=make(chanstruct{},1)//thecapacitymustbeone counter:=0 increase:=func(){ mutex<-?struct{}{}?//?lock ????????counter++ ????????<-mutex?//?unlock ????} ????increase1000?:=?func(done?chan<-?struct{})?{ ????????for?i?:=?0;?i?

關閉 channel

關閉不再需要使用的 channel 并不是必須的。跟其他資源比如打開的文件、socket 連接不一樣，這類資源使用完后不關閉后會造成句柄泄露，channel 使用完后不關閉也沒有關系，channel 沒有被任何協(xié)程用到后最終會被 GC 回收。關閉 channel 一般是用來通知其他協(xié)程某個任務已經完成了。golang 也沒有直接提供判斷 channel 是否已經關閉的接口，雖然可以用其他不太優(yōu)雅的方式自己實現一個：

funcisClosed(chchanint)bool{ select{ case<-ch: ????????return?true ????default: ????} ????return?false }不過實現一個這樣的接口也沒什么必要。因為就算通過?isClosed()?得到當前 channel 當前還未關閉，如果試圖往 channel 里寫數據，仍然可能會發(fā)生 panic ，因為在調用?isClosed()?后，其他協(xié)程可能已經把 channel 關閉了。關閉 channel 時應該注意以下準則：

不要在讀取端關閉 channel ，因為寫入端無法知道 channel 是否已經關閉，往已關閉的 channel 寫數據會 panic ；

有多個寫入端時，不要再寫入端關閉 channle ，因為其他寫入端無法知道 channel 是否已經關閉，關閉已經關閉的 channel 會發(fā)生 panic ；

如果只有一個寫入端，可以在這個寫入端放心關閉 channel 。

關閉 channel 粗暴一點的做法是隨意關閉，如果產生了 panic 就用 recover 避免進程掛掉。稍好一點的方案是使用標準庫的sync包來做關閉 channel 時的協(xié)程同步，不過使用起來也稍微復雜些。下面介紹一種優(yōu)雅些的做法。

一寫多讀

這種場景下這個唯一的寫入端可以關閉 channel 用來通知讀取端所有數據都已經寫入完成了。讀取端只需要用for range把 channel 中數據遍歷完就可以了，當 channel 關閉時，for range仍然會將 channel 緩沖中的數據全部遍歷完然后再退出循環(huán)：

packagemain import( "fmt" "sync" ) funcmain(){ wg:=&sync.WaitGroup{} ch:=make(chanint,100) send:=func(){ fori:=0;i

多寫一讀

這種場景下雖然可以用sync.Once來解決多個寫入端重復關閉 channel 的問題，但更優(yōu)雅的辦法設置一個額外的 channel ，由讀取端通過關閉來通知寫入端任務完成不要再繼續(xù)再寫入數據了：

packagemain import( "fmt" "sync" ) funcmain(){ wg:=&sync.WaitGroup{} ch:=make(chanint,100) done:=make(chanstruct{}) send:=func(idint){ deferwg.Done() fori:=0;;i++{ select{ case<-done: ????????????????//?get?exit?signal ????????????????fmt.Printf("sender?#%d?exit ",?id) ????????????????return ????????????case?ch?<-?id*1000?+?i: ????????????} ????????} ????} ????recv?:=?func()?{ ????????count?:=?0 ????????for?i?:=?range?ch?{ ????????????fmt.Printf("receiver?get?%d ",?i) ????????????count++ ????????????if?count?>=1000{ //signalrecvingfinish close(done) return } } } wg.Add(3) gosend(0) gosend(1) gosend(2) recv() wg.Wait() }

多寫多讀

這種場景稍微復雜，和上面的例子一樣，也需要設置一個額外 channel 用來通知多個寫入端和讀取端。另外需要起一個額外的協(xié)程來通過關閉這個 channel 來廣播通知：

packagemain import( "fmt" "sync" "time" ) funcmain(){ wg:=&sync.WaitGroup{} ch:=make(chanint,100) done:=make(chanstruct{}) send:=func(idint){ deferwg.Done() fori:=0;;i++{ select{ case<-done: ????????????????//?get?exit?signal ????????????????fmt.Printf("sender?#%d?exit ",?id) ????????????????return ????????????case?ch?<-?id*1000?+?i: ????????????} ????????} ????} ????recv?:=?func(id?int)?{ ????????defer?wg.Done() ????????for?{ ????????????select?{ ????????????case?<-done: ????????????????//?get?exit?signal ????????????????fmt.Printf("receiver?#%d?exit ",?id) ????????????????return ????????????case?i?:=?<-ch: ????????????????fmt.Printf("receiver?#%d?get?%d ",?id,?i) ????????????????time.Sleep(time.Millisecond) ????????????} ????????} ????} ????wg.Add(6) ????go?send(0) ????go?send(1) ????go?send(2) ????go?recv(0) ????go?recv(1) ????go?recv(2) ????time.Sleep(time.Second) ????//?signal?finish ????close(done) ????//?wait?all?sender?and?receiver?exit ????wg.Wait() }

總結

channle 作為 golang 最重要的特性，用起來還是比較爽的。傳統(tǒng)的 C 里要實現類型的功能的話，一般需要用到 socket 或者 FIFO 來實現，另外還要考慮數據包的完整性與并發(fā)沖突的問題，channel 則屏蔽了這些底層細節(jié)，使用者只需要考慮讀寫就可以了。channel 是引用類型，了解一下 channel 底層的機制對更好的使用 channel 還是很用必要的。

雖然操作原語簡單，但涉及到阻塞的問題，使用不當可能會造成死鎖或者無限制的協(xié)程創(chuàng)建最終導致進程掛掉。 channel 除在可以用來在協(xié)程之間通信外，其阻塞和喚醒協(xié)程的特性也可以用作協(xié)程之間的同步機制，文中也用示例簡單介紹了這種場景下的用法。

關閉 channel 并不是必須的，只要沒有協(xié)程沒用引用 channel ，最終會被 GC 清理。所以使用的時候要特別注意，不要讓協(xié)程阻塞在 channel 上，這種情況很難檢測到，而且會造成 channel 和阻塞在 channel 的協(xié)程占有的資源無法被 GC 清理最終導致內存泄露。

channle 方便 golang 程序使用 CSP 的編程范形，但是 golang 是一種多范形的編程語言，golang 也支持傳統(tǒng)的通過共享內存來通信的編程方式。終極的原則是根據場景選擇合適的編程范型，不要因為 channel 好用而濫用 CSP 。

轉自：http://litang.me/post/golang-channel/

編輯：jq