Golang 併發簡介

併發概要

隨着多核CPU的普及, 爲了更快的處理任務, 出現了各類併發編程的模型, 主要有如下幾種:

模型名稱	優勢	缺點
多進程	簡單, 隔離性好, 進程間幾乎無影響	開銷最大
多線程	目前使用最多的方式, 開銷比多進程小	高併發模式下, 效率會有影響
異步	相比多線程而言, 能夠減小線程的數量	編碼要求高, 須要對流程分割合理
協程	用戶態線程, 不須要操做系統來調度, 因此輕量, 開銷極小	須要語言支持

協程介紹

協程是個抽象的概念, 能夠映射到到操做系統層面的進程, 線程等概念.
因爲協程是用戶態的線程, 不用操做系統來調度, 因此不受操做系統的限制, 能夠輕鬆的建立百萬個, 所以也被稱爲 "輕量級線程".

在 golang 中, 協程不是由庫實現的, 而是受語言級別支持的, 所以, 在 golang 中, 使用協程很是方便.
下面經過例子演示在 golang 中, 如何使用協程來完成併發操做.

golang 併發

實現方式

golang 中, 經過 go 關鍵字能夠很是簡單的啓動一個協程, 幾乎沒有什麼學習成本.
固然併發編程中固有的業務上的困難依然存在(好比並發時的同步, 超時等), 可是 golang 在語言級別給咱們提供了優雅簡潔的解決這些問題的途徑.

理解了 golang 中協程的使用, 會給咱們寫併發程序時帶來極大的便利.
首先以一個簡單的例子開始 golang 的併發編程.

package main

import (
     "fmt"
     "time"
)

func main() {
     for i := 0; i < 10; i++ {
             go sum(i, i+10)
     }

     time.Sleep(time.Second * 5)
}

func sum(start, end int) int {
     var sum int = 0
     for i := start; i < end; i++ {
             sum += i
     }

     fmt.Printf("Sum from %d to %d is %d\n", start, end, sum)
     return sum
}

執行結果以下: (同時啓動10個協程作累加運算, 10個協程的執行順序可能會不同)

$ go run main.go
Sum from 0 to 10 is 45
Sum from 6 to 16 is 105
Sum from 7 to 17 is 115
Sum from 2 to 12 is 65
Sum from 8 to 18 is 125
Sum from 1 to 11 is 55
Sum from 9 to 19 is 135
Sum from 3 to 13 is 75
Sum from 4 to 14 is 85
Sum from 5 to 15 is 95

經過 go 關鍵字啓動協程以後, 主進程並不會等待協程的執行, 而是繼續執行直至結束.
本例中, 若是沒有 time.Sleep(time.Second * 5) 等待5秒的話, 那麼主進程不會等待那10個協程的運行結果, 直接就結束了.
主進程結束也會致使那10個協程的執行中斷, 因此, 若是去掉 time.Sleep 這行代碼, 可能屏幕上什麼顯示也沒有.

簡單示例

實際使用協程時, 咱們通常會等待全部協程執行完成(或者超時)後, 纔會結束主進程, 可是不會用 time.Sleep 這種方式,
由於主進程並不知道協程何時會結束, 無法設置等待時間.

這時, 就看出 golang 中的 channel 機制所帶來的好處了. 下面用 channel 來改造上面的 time.Sleep

package main

import "fmt"

func main() {
     var ch = make(chan string)

     for i := 0; i < 10; i++ {
             go sum(i, i+10, ch)
     }

     for i := 0; i < 10; i++ {
             fmt.Print(<-ch)
     }
}

func sum(start, end int, ch chan string) {

     var sum int = 0
     for i := start; i < end; i++ {
             sum += i
     }

     ch <- fmt.Sprintf("Sum from %d to %d is %d\n", start, end, sum)
}

程序執行結果和上面同樣, 由於是併發的緣故, 可能輸出的 sum 順序可能會不同.

$ go run main.go
Sum from 9 to 19 is 135
Sum from 0 to 10 is 45
Sum from 5 to 15 is 95
Sum from 6 to 16 is 105
Sum from 7 to 17 is 115
Sum from 2 to 12 is 65
Sum from 8 to 18 is 125
Sum from 3 to 13 is 75
Sum from 1 to 11 is 55
Sum from 4 to 14 is 85

golang 的 chan 能夠是任意類型的, 上面的例子中定義的是 string 型.
從上面的程序能夠看出, 往 chan 中寫入數據以後, 協程會阻塞在那裏, 直到在某個地方將 chan 中的值讀取出來, 協程纔會繼續運行下去.

上面的例子中, 咱們啓動了10個協程, 每一個協程都往 chan 中寫入了一個字符串, 而後在 main 函數中, 依次讀取 chan 中的字符串, 並在屏幕上打印出來.
經過 golang 中的 chan, 不只實現了主進程 和 協程之間的通訊, 並且不用像 time.Sleep 那樣不可控(由於你不知道要 Sleep 多長時間).

併發時的緩衝

上面的例子中, 全部協程使用的是同一個 chan, chan 的容量默認只有 1, 當某個協程向 chan 中寫入數據時, 其餘協程再次向 chan 中寫入數據時, 實際上是阻塞的.
等到 chan 中的數據被讀出以後, 纔會再次讓某個其餘協程寫入, 由於每一個協程都執行的很是快, 因此看不出來.

改造下上面的例子, 加入些 Sleep 代碼, 延長每一個協程的執行時間, 咱們就能夠看出問題, 代碼以下:

package main

import (
     "fmt"
     "time"
)

func main() {
     var ch = make(chan string)

     for i := 0; i < 5; i++ {
             go sum(i, i+10, ch)
     }

     for i := 0; i < 10; i++ {
             time.Sleep(time.Second * 1)
             fmt.Print(<-ch)
     }
}

func sum(start, end int, ch chan string) int {
     ch <- fmt.Sprintf("Sum from %d to %d is starting at %s\n", start, end, time.Now().String())
     var sum int = 0
     for i := start; i < end; i++ {
             sum += i
     }
     time.Sleep(time.Second * 10)
     ch <- fmt.Sprintf("Sum from %d to %d is %d at %s\n", start, end, sum, time.Now().String())
     return sum
}

執行結果以下:

$ go run main.go
Sum from 4 to 14 is starting at 2015-10-13 13:59:56.025633342 +0800 CST
Sum from 3 to 13 is starting at 2015-10-13 13:59:56.025608644 +0800 CST
Sum from 0 to 10 is starting at 2015-10-13 13:59:56.025508327 +0800 CST
Sum from 2 to 12 is starting at 2015-10-13 13:59:56.025574486 +0800 CST
Sum from 1 to 11 is starting at 2015-10-13 13:59:56.025593711 +0800 CST
Sum from 4 to 14 is 85 at 2015-10-13 14:00:07.030611465 +0800 CST
Sum from 3 to 13 is 75 at 2015-10-13 14:00:08.031926629 +0800 CST
Sum from 0 to 10 is 45 at 2015-10-13 14:00:09.036724803 +0800 CST
Sum from 2 to 12 is 65 at 2015-10-13 14:00:10.038125044 +0800 CST
Sum from 1 to 11 is 55 at 2015-10-13 14:00:11.040366206 +0800 CST

爲了演示 chan 的阻塞狀況, 上面的代碼中特地加了一些 time.Sleep 函數.

• 每一個執行 Sum 函數的協程都會運行 10 秒
• main函數中每隔 1 秒讀一次 chan 中的數據

從打印結果咱們能夠看出, 全部協程幾乎是同一時間開始的, 說明了協程確實是併發的.
其中, 最快的協程(Sum from 4 to 14…)執行了 11 秒左右, 爲何是 11 秒左右呢?
說明它阻塞在了 Sum 函數中的第一行上, 等了 1 秒以後, main 函數開始讀出 chan 中數據後才繼續運行.
它自身運行須要 10 秒, 加上等待的 1 秒, 正好 11 秒左右.

最慢的協程執行了 15 秒左右, 這個也很好理解, 總共啓動了 5 個協程, main 函數每隔 1 秒 讀出一次 chan, 最慢的協程等待了 5 秒,
再加上自身執行了 10 秒, 因此一共 15 秒左右.

到這裏, 咱們很天然會想到可否增長 chan 的容量, 從而使得每一個協程儘快執行, 完成本身的操做, 而不用等待, 消除因爲 main 函數的處理所帶來的瓶頸呢?
答案是固然能夠, 並且在 golang 中實現還很簡單, 只要在建立 chan 時, 指定 chan 的容量就行.

package main

import (
     "fmt"
     "time"
)

func main() {
     var ch = make(chan string, 10)

     for i := 0; i < 5; i++ {
             go sum(i, i+10, ch)
     }

     for i := 0; i < 10; i++ {
             time.Sleep(time.Second * 1)
             fmt.Print(<-ch)
     }
}

func sum(start, end int, ch chan string) int {
     ch <- fmt.Sprintf("Sum from %d to %d is starting at %s\n", start, end, time.Now().String())
     var sum int = 0
     for i := start; i < end; i++ {
             sum += i
     }
     time.Sleep(time.Second * 10)
     ch <- fmt.Sprintf("Sum from %d to %d is %d at %s\n", start, end, sum, time.Now().String())
     return sum
}

執行結果以下:

$ go run main.go
Sum from 0 to 10 is starting at 2015-10-13 14:22:14.64534265 +0800 CST
Sum from 2 to 12 is starting at 2015-10-13 14:22:14.645382961 +0800 CST
Sum from 3 to 13 is starting at 2015-10-13 14:22:14.645408947 +0800 CST
Sum from 4 to 14 is starting at 2015-10-13 14:22:14.645417257 +0800 CST
Sum from 1 to 11 is starting at 2015-10-13 14:22:14.645427028 +0800 CST
Sum from 1 to 11 is 55 at 2015-10-13 14:22:24.6461138 +0800 CST
Sum from 3 to 13 is 75 at 2015-10-13 14:22:24.646330223 +0800 CST
Sum from 2 to 12 is 65 at 2015-10-13 14:22:24.646325521 +0800 CST
Sum from 4 to 14 is 85 at 2015-10-13 14:22:24.646343061 +0800 CST
Sum from 0 to 10 is 45 at 2015-10-13 14:22:24.64634674 +0800 CST

從執行結果能夠看出, 全部協程幾乎都是 10秒完成的. 因此在使用協程時, 記住能夠經過使用緩存來進一步提升併發性.

併發時的超時

併發編程, 因爲不能確保每一個協程都能及時響應, 有時候協程長時間沒有響應, 主進程不可能一直等待, 這時候就須要超時機制.
在 golang 中, 實現超時機制也很簡單.

package main

import (
     "fmt"
     "time"
)

func main() {
     var ch = make(chan string, 1)
     var timeout = make(chan bool, 1)

     go sum(1, 10, ch)
     go func() {
             time.Sleep(time.Second * 5) // 5 秒超時
             timeout <- true
     }()

     select {
     case sum := <-ch:
             fmt.Print(sum)
     case <-timeout:
             fmt.Println("Sorry, TIMEOUT!")
     }
}

func sum(start, end int, ch chan string) int {
     var sum int = 0
     for i := start; i < end; i++ {
             sum += i
     }
     time.Sleep(time.Second * 10)
     ch <- fmt.Sprintf("Sum from %d to %d is %d\n", start, end, sum)
     return sum
}

經過一個匿名函數來控制超時, 而後同時啓動 計算 sum 的協程和timeout協程, 在 select 中看誰先結束,
若是 timeout 結束後, 計算 sum 的協程尚未結束的話, 就會進入超時處理.

上例中, timeout 只有5秒, sum協程會執行10秒, 因此執行結果以下:

$ go run main.go
Sorry, TIMEOUT!

修改 time.Sleep(time.Second * 5) 爲 time.Sleep(time.Second * 15) 的話, 就會看到 sum 協程的執行結果