Golang 中的并發限制與逾時控制

前言

上回在 用 Go 寫一個輕量級的 ssh 批量操作工具 裡提及過，我們做 Golang 并發的時候要對并發進行限制，對 goroutine 的執行要有逾時控制。那會沒有細說，這裡展開讨論一下。

以下示例代碼全部可以直接在 The Go Playground 上運作測試：

并發

我們先來跑一個簡單的并發看看

1package main
 2
 3import (
 4    "fmt"
 5    "time"
 6)
 7
 8func run(task_id, sleeptime int, ch chan string) {
 9
10    time.Sleep(time.Duration(sleeptime) * time.Second)
11    ch <- fmt.Sprintf("task id %d , sleep %d second", task_id, sleeptime)
12    return
13}
14
15func main() {
16    input := []int{3, 2, 1}
17    ch := make(chan string)
18    startTime := time.Now()
19    fmt.Println("Multirun start")
20    for i, sleeptime := range input {
21        go run(i, sleeptime, ch)
22    }
23
24    for range input {
25        fmt.Println(<-ch)
26    }
27
28    endTime := time.Now()
29    fmt.Printf("Multissh finished. Process time %s. Number of tasks is %d", endTime.Sub(startTime), len(input))
30}           

函數 run() 接受輸入的參數，sleep 若幹秒。然後通過 go 關鍵字并發執行，通過 channel 傳回結果。

channel 顧名思義，他就是 goroutine 之間通信的“管道"。管道中的資料流通，實際上是 goroutine 之間的一種記憶體共享。我們通過他可以在 goroutine 之間互動資料。

1ch <- xxx // 向 channel 寫入資料
2<- ch // 從 channel 中讀取資料           

channel 分為無緩沖(unbuffered)和緩沖(buffered)兩種。例如剛才我們通過如下方式建立了一個無緩沖的 channel。

1ch := make(chan string)           

channel 的緩沖，我們一會再說，先看看剛才看看執行的結果。

1Multirun start
2task id 2 , sleep 1 second
3task id 1 , sleep 2 second
4task id 0 , sleep 3 second
5Multissh finished. Process time 3s. Number of tasks is 3
6Program exited.           

三個 goroutine `分别 sleep 了 3，2，1秒。但總耗時隻有 3 秒。是以并發生效了，go 的并發就是這麼簡單。

按序傳回

剛才的示例中，我執行任務的順序是 0，1，2。但是從 channel 中傳回的順序卻是 2，1，0。這很好了解，因為 task 2 執行的最快嘛，是以先傳回了進入了 channel，task 1 次之，task 0 最慢。

如果我們希望按照任務執行的順序依次傳回資料呢？可以通過一個 channel 數組（好吧，應該叫切片）來做，比如這樣

1package main
 2
 3import (
 4    "fmt"
 5    "time"
 6)
 7
 8func run(task_id, sleeptime int, ch chan string) {
 9
10    time.Sleep(time.Duration(sleeptime) * time.Second)
11    ch <- fmt.Sprintf("task id %d , sleep %d second", task_id, sleeptime)
12    return
13}
14
15func main() {
16    input := []int{3, 2, 1}
17    chs := make([]chan string, len(input))
18    startTime := time.Now()
19    fmt.Println("Multirun start")
20    for i, sleeptime := range input {
21        chs[i] = make(chan string)
22        go run(i, sleeptime, chs[i])
23    }
24
25    for _, ch := range chs {
26        fmt.Println(<-ch)
27    }
28
29    endTime := time.Now()
30    fmt.Printf("Multissh finished. Process time %s. Number of tasks is %d", endTime.Sub(startTime), len(input))
31}           

運作結果，現在輸出的次序和輸入的次序一緻了。

1Multirun start
2task id 0 , sleep 3 second
3task id 1 , sleep 2 second
4task id 2 , sleep 1 second
5Multissh finished. Process time 3s. Number of tasks is 3
6Program exited.           

逾時控制

剛才的例子裡我們沒有考慮逾時。然而如果某個 goroutine 運作時間太長了，那很肯定會拖累主 goroutine 被阻塞住，整個程式就挂起在那兒了。是以我們需要有逾時的控制。

通常我們可以通過select + time.After 來進行逾時檢查，例如這樣，我們增加一個函數 Run() ，在 Run() 中執行 go run() 。并通過 select + time.After 進行逾時判斷。

1package main
 2
 3import (
 4    "fmt"
 5    "time"
 6)
 7
 8func Run(task_id, sleeptime, timeout int, ch chan string) {
 9    ch_run := make(chan string)
10    go run(task_id, sleeptime, ch_run)
11    select {
12    case re := <-ch_run:
13        ch <- re
14    case <-time.After(time.Duration(timeout) * time.Second):
15        re := fmt.Sprintf("task id %d , timeout", task_id)
16        ch <- re
17    }
18}
19
20func run(task_id, sleeptime int, ch chan string) {
21
22    time.Sleep(time.Duration(sleeptime) * time.Second)
23    ch <- fmt.Sprintf("task id %d , sleep %d second", task_id, sleeptime)
24    return
25}
26
27func main() {
28    input := []int{3, 2, 1}
29    timeout := 2
30    chs := make([]chan string, len(input))
31    startTime := time.Now()
32    fmt.Println("Multirun start")
33    for i, sleeptime := range input {
34        chs[i] = make(chan string)
35        go Run(i, sleeptime, timeout, chs[i])
36    }
37
38    for _, ch := range chs {
39        fmt.Println(<-ch)
40    }
41    endTime := time.Now()
42    fmt.Printf("Multissh finished. Process time %s. Number of task is %d", endTime.Sub(startTime), len(input))
43}           

運作結果，task 0 和 task 1 已然逾時

1Multirun start
2task id 0 , timeout
3task id 1 , timeout
4tasi id 2 , sleep 1 second
5Multissh finished. Process time 2s. Number of task is 3
6Program exited.           

并發限制

如果任務數量太多，不加以限制的并發開啟 goroutine 的話，可能會過多的占用資源，伺服器可能會爆炸。是以實際環境中并發限制也是一定要做的。

一種常見的做法就是利用 channel 的緩沖機制——開始的時候我們提到過的那個。

我們分别建立一個帶緩沖和不帶緩沖的 channel 看看

1ch := make(chan string) // 這是一個無緩沖的 channel，或者說緩沖區長度是 0
2ch := make(chan string, 1) // 這是一個帶緩沖的 channel, 緩沖區長度是 1            

這兩者的差別在于，如果 channel 沒有緩沖，或者緩沖區滿了。goroutine 會自動阻塞，直到 channel 裡的資料被讀走為止。舉個例子

1package main
 2
 3import (
 4    "fmt"
 5)
 6
 7func main() {
 8    ch := make(chan string)
 9    ch <- "123"
10    fmt.Println(<-ch)
11}            

這段代碼執行将報錯

1fatal error: all goroutines are asleep - deadlock!
2
3goroutine 1 [chan send]:
4main.main()
5    /tmp/sandbox531498664/main.go:9 +0x60
6
7Program exited.           

這是因為我們建立的 ch 是一個無緩沖的 channel。是以在執行到 ch<-"123",這個 goroutine 就阻塞了，後面的 fmt.Println(<-ch) 沒有辦法得到執行。是以将會報 deadlock 錯誤。

如果我們改成這樣，程式就可以執行

1package main
 2
 3import (
 4    "fmt"
 5)
 6
 7func main() {
 8    ch := make(chan string, 1)
 9    ch <- "123"
10    fmt.Println(<-ch)
11}           

執行

1123
2
3Program exited.           

如果我們改成這樣

1package main
 2
 3import (
 4    "fmt"
 5)
 6
 7func main() {
 8    ch := make(chan string, 1)
 9    ch <- "123"
10    ch <- "123"
11    fmt.Println(<-ch)
12    fmt.Println(<-ch)
13}           

盡管讀取了兩次 channel，但是程式還是會死鎖，因為緩沖區滿了，goroutine 阻塞挂起。第二個 ch<- "123" 是沒有辦法寫入的。

1fatal error: all goroutines are asleep - deadlock!
2
3goroutine 1 [chan send]:
4main.main()
5    /tmp/sandbox642690323/main.go:10 +0x80
6
7Program exited.           

是以，利用 channel 的緩沖設定，我們就可以來實作并發的限制。我們隻要在執行并發的同時，往一個帶有緩沖的 channel 裡寫入點東西（随便寫啥，内容不重要）。讓并發的 goroutine 在執行完成後把這個 channel 裡的東西給讀走。這樣整個并發的數量就講控制在這個 channel 的緩沖區大小上。

比如我們可以用一個 bool 類型的帶緩沖 channel 作為并發限制的計數器。

1    chLimit := make(chan bool, 1)           

然後在并發執行的地方，每建立一個新的 goroutine，都往 chLimit 裡塞個東西。

1for i, sleeptime := range input {
2        chs[i] = make(chan string, 1)
3        chLimit <- true
4        go limitFunc(chLimit, chs[i], i, sleeptime, timeout)
5    }           

這裡通過 go 關鍵字并發執行的是新構造的函數。他在執行完原來的 Run() 後，會把 chLimit 的緩沖區裡給消費掉一個。

1limitFunc := func(chLimit chan bool, ch chan string, task_id, sleeptime, timeout int) {
2        Run(task_id, sleeptime, timeout, ch)
3        <-chLimit
4    }           

這樣一來，當建立的 goroutine 數量到達 chLimit 的緩沖區上限後。主 goroutine 就挂起阻塞了，直到這些 goroutine 執行完畢，消費掉了 chLimit 緩沖區中的資料，程式才會繼續建立新的 goroutine。我們并發數量限制的目的也就達到了。

以下是完整代碼

1package main
 2
 3import (
 4    "fmt"
 5    "time"
 6)
 7
 8func Run(task_id, sleeptime, timeout int, ch chan string) {
 9    ch_run := make(chan string)
10    go run(task_id, sleeptime, ch_run)
11    select {
12    case re := <-ch_run:
13        ch <- re
14    case <-time.After(time.Duration(timeout) * time.Second):
15        re := fmt.Sprintf("task id %d , timeout", task_id)
16        ch <- re
17    }
18}
19
20func run(task_id, sleeptime int, ch chan string) {
21
22    time.Sleep(time.Duration(sleeptime) * time.Second)
23    ch <- fmt.Sprintf("task id %d , sleep %d second", task_id, sleeptime)
24    return
25}
26
27func main() {
28    input := []int{3, 2, 1}
29    timeout := 2
30    chLimit := make(chan bool, 1)
31    chs := make([]chan string, len(input))
32    limitFunc := func(chLimit chan bool, ch chan string, task_id, sleeptime, timeout int) {
33        Run(task_id, sleeptime, timeout, ch)
34        <-chLimit
35    }
36    startTime := time.Now()
37    fmt.Println("Multirun start")
38    for i, sleeptime := range input {
39        chs[i] = make(chan string, 1)
40        chLimit <- true
41        go limitFunc(chLimit, chs[i], i, sleeptime, timeout)
42    }
43
44    for _, ch := range chs {
45        fmt.Println(<-ch)
46    }
47    endTime := time.Now()
48    fmt.Printf("Multissh finished. Process time %s. Number of task is %d", endTime.Sub(startTime), len(input))
49}           

運作結果

1Multirun start
2task id 0 , timeout
3task id 1 , timeout
4task id 2 , sleep 1 second
5Multissh finished. Process time 5s. Number of task is 3
6Program exited.           

chLimit 的緩沖是 1。task 0 和 task 1 耗時 2 秒逾時。task 2 耗時 1 秒。總耗時 5 秒。并發限制生效了。

如果我們修改并發限制為 2

1chLimit := make(chan bool, 2)           

1Multirun start
2task id 0 , timeout
3task id 1 , timeout
4task id 2 , sleep 1 second
5Multissh finished. Process time 3s. Number of task is 3
6Program exited.           

task 0 , task 1 并發執行，耗時 2秒。task 2 耗時 1秒。總耗時 3 秒。符合預期。

有沒有注意到代碼裡有個地方和之前不同。這裡，用了一個帶緩沖的 channel

1chs[i] = make(chan string, 1)           

還記得上面的例子麼。如果 channel 不帶緩沖，那麼直到他被消費掉之前，這個 goroutine 都會被阻塞挂起。

然而如果這裡的并發限制，也就是 chLimit 生效阻塞了主 goroutine，那麼後面消費這些資料的代碼并不會執行到。。。于是就 deadlock 拉！

1  for _, ch := range chs {
2        fmt.Println(<-ch)
3    }           

是以給他一個緩沖就好了。

參考文獻

從Deadlock報錯了解Go channel機制（一）

golang-what-is-channel-buffer-size

golang-using-timeouts-with-channels

原文釋出時間為：2018-12-6

本文作者：Golang語言社群

本文來自雲栖社群合作夥伴“

Golang語言社群

”，了解相關資訊可以關注“Golangweb”微信公衆号