講透Go中的并發接收控制結構Select
select 與 switch
讓我們來復習一下switch語句，在switch語句中，會逐個匹配case語句(可以是值也可以是表達式)，一個一個的判斷過去，直到有符合的語句存在，執行匹配的語句內容后跳出switch。
- func demo(number int){
- switch{
- case number >= 90:
- fmt.Println("優秀")
- default:
- fmt.Println("太搓了")
- }
- }
而 select 用于處理通道，它的語法與 switch 非常類似。每個 case 語句里必須是一個 channel 操作。它既可以用于 channel 的數據接收，也可以用于 channel 的數據發送。
- func foo() {
- chanInt := make(chan int)
- defer close(chanInt)
- go func() {
- select {
- case data, ok := <-chanInt:
- if ok {
- fmt.Println(data)
- }
- default:
- fmt.Println("全部阻塞")
- }
- }()
- chanInt <- 1
- }
輸出1
- 這是一個簡單的接收發送模型
- 如果 select 的多個分支都滿足條件，則會隨機的選取其中一個滿足條件的分支。
- 第 6 行加上 ok 是因為上一節講過，如果不加會導致通道關閉時收到零值
- 回憶之前的知識，讓接收和發送在不同的goroutine里，否則會死鎖
這個程序存在什么問題?
假如發送太慢，所有case都處于阻塞狀態，會直接執行default的內容。這里加一行sleep試試。
- func bar() {
- chanInt := make(chan int)
- defer close(chanInt)
- go func() {
- ....
- }()
- time.Sleep(time.Second)
- chanInt <- 1
- }
- 倒數第二行加了sleep 1 秒，導致select語句提前結束
- 猜測一下會輸出全部阻塞嗎?
- 全部阻塞
- fatal error: all goroutines are asleep - deadlock!
- goroutine 1 [chan send]:
- main.bar()
是會輸出全部阻塞的。
因為接收執行完了，退出了goroutine，而發送才剛剛執行到，沒有與其匹配的接收，故死鎖。
正確的做法是把接收套在循環里面。
- func baz() {
- chanInt := make(chan int)
- defer close(chanInt)
- go func() {
- for {
- select {
- ...
- }
- }
- }()
- chanInt <- 1
- }
- 不再死鎖了
- 假如程序不停止，會出現一個泄露的goroutine，永遠的在for循環中無法跳出，此時引入下一節的內容
通知機制
Go 語言總是簡單和靈活的，雖然沒有針對提供專門的機制來處理退出，但我們可以自己組合
- func main() {
- chanInt, done := make(chan int), make(chan struct{})
- defer close(chanInt)
- defer close(done)
- go func() {
- for {
- select {
- case <-chanInt:
- case <-done:
- break
- }
- }
- }()
- done <- struct{}{}
- }
沒有給chanInt發送任何東西，按理說會阻塞，導致goroutine泄露
但可以使用額外的通道完成協程的退出控制
這種方式還可以做到周期性處理任務，下一節我們再詳細講解
case 的并發性
case是有并發屬性的，比如兩次輸入，分別等待 1、2 秒，再進行兩次讀取，會花 3 秒時間嗎?
- func main() {
- c1,c2 := make(chan string), make(chan string)
- close(c1)
- close(c2)
- go func() {
- time.Sleep(time.Second * 1)
- c1 <- "one"
- }()
- go func() {
- time.Sleep(time.Second * 2)
- c2 <- "two"
- }()
- start := time.Now() // 獲取當前時間
- for i := 0; i < 2; i++ {
- select {
- case <-c1:
- case <-c2:
- }
- }
- elapsed := time.Since(start)
- // 這里沒有用到3秒，為什么？
- fmt.Println("該函數執行完成耗時：", elapsed)
- }
以上代碼先初始化兩個 channel c1 和 c2，然后開啟兩個 goroutine 分別往 c1 和 c2 寫入數據，再通過 select 監聽兩個 channel，從中讀取數據并輸出。
運行結果如下：
- $ go run channel.go
- received one
- received two
- 該函數執行完成耗時：2.004695535s
這充分說明case是并發的，但要注意此處的并發是 case 對channel阻塞做出的特殊處理。
case并發的原理
假如case后左邊和右邊跟了函數，會執行函數，我們來探索一下。
定義A、B函數，作用相同
- func A() int {
- fmt.Println("start A")
- time.Sleep(1 * time.Second)
- fmt.Println("end A")
- return 1
- }
定義函數lee，請問該函數執行完成耗時多少呢?
- func lee() {
- ch, done := make(chan int), make(chan struct{})
- defer close(ch)
- go func() {
- select {
- case ch <- A():
- case ch <- B():
- case <-done:
- }
- }()
- done <- struct{}{}
- }
答案是 2 秒
- start A
- end A
- start B
- end B
- main.leespend time: 2.003504395s
- select 掃描是從左到右從上到下的，按這個順序先求值，如果是函數會先執行函數。
- 然后立馬判斷是否可以立即執行(這里是指 case 是否會因為執行而阻塞)。
- 所以兩個函數都會進入，而且是先進入 A 再進入 B，兩個函數都會執行完，所以等待時間會累計。
如果都不會阻塞，此時就會使用一個偽隨機的算法，去選中一個 case，只要選中了其他就被放棄了。
超時控制
我們來模擬一個更真實點的例子，讓程序一段時間超時退出。
定義一個結構體
- type Worker struct {
- stream <-chan int //處理
- timeout time.Duration //超時
- done chan struct{} //結束信號
- }
定義初始化函數
- func NewWorker(stream <-chan int, timeout int) *Worker {
- return &Worker{
- stream: stream,
- timeout: time.Duration(timeout) * time.Second,
- done: make(chan struct{}),
- }
- }
定義超時處理函數
- func (w *Worker) afterTimeStop() {
- go func() {
- time.Sleep(w.timeout)
- w.done <- struct{}{}
- }()
- }
- 超過時間發送結束信號
接收數據并處理函數
- func (w *Worker) Start() {
- w.afterTimeStop()
- for {
- select {
- case data, ok := <-w.stream:
- if !ok {
- return
- }
- fmt.Println(data)
- case <-w.done:
- close(w.done)
- return
- }
- }
- }
- 收到結束信號關閉函數
- 這樣的方法就可以讓程序在等待 1 秒后繼續執行，而不會因為 ch 讀取等待而導致程序停滯。
- func main() {
- stream := make(chan int)
- defer close(stream)
- w := NewWorker(stream, 3)
- w.Start()
- }
實際 3 秒到程序運行結束。
這種方式巧妙地實現了超時處理機制，這種方法不僅簡單，在實際項目開發中也是非常實用的。
小結
本節介紹了select的用法以及包含的陷阱，我們學會了
- case是并發的
- case只針對通道傳輸阻塞做特殊處理，如果有計算將會先進行計算
- 掃描是從左到右從上到下的，按這個順序先求值，如果是函數會先執行函數。如果函數運行時間長，時間會累計
- 在case全部阻塞時，會執行default中的內容
- 可使用結束信號，讓select退出
- 延時發送結束信號可以實現超時自動退出的功能
問題：為什么w.stream沒有程序向他發送數據，卻沒有死鎖呢?
本節源碼位置 https://github.com/golang-minibear2333/golang/blob/master/4.concurrent/4.5-select”
本文轉載自微信公眾號「機智的程序員小熊」，可以通過以下二維碼關注。轉載本文請聯系機智的程序員小熊公眾號。
