理解golang調度之三：併發

時間 2019-11-08

標籤理解 golang 調度之三併發欄目 Go 简体版

原文原文鏈接

前言

這一部分有三篇文章，主要是講解go調度器的一些內容html

三篇文章分別是:git

簡介

當我在解決一個問題尤爲是新問題的時候，我開始不會去考慮併發(concurrency)是否合適。我首先會去找一系列的解決方式而後確保它有效。而後在可讀性和技術方案評估以後，我會開始去考慮併發是否實際合理。有些時候併發的好處是顯而易見的，可是有時候並非很明顯。github

第一篇文章，我解釋了OS調度器的相關內容，我以爲這部分對於你寫多線程代碼很重要。第二篇裏，我講解了一些Go調度器的一些內容，這部分對於你理解和寫go的併發代碼頗有幫助。在這篇文章裏，我會在OS和Go調度器層面讓你去深層次的理解併發究竟是什麼。golang

這部份內容的目標是：算法

你的工做負載(workloads)使用併發是否合適，爲此提供一些指導建議
不一樣工做負載的含義，並針對其做出相應的工程方面的決策。

什麼是併發

併發的含義就是無序的執行。給你一系列的指令，去找到一個方式能夠無序執行並且和有序執行產生一樣的結果。這個問題在你面前，顯而易見的是無序執行會增長一些足夠的性能增益在計算了複雜性成本以後，可是你可能會以爲無序執行是不可能的甚至是沒有意義的。bash

你也要清楚一點，併發和並行是不同的。並行是在相同時間內同時執行兩個或兩個以上的指令，這和併發的概念不同。網絡

圖3.1

圖3.1裏，你看到主機上有兩個邏輯處理器。每一個都有他們單獨的OS線程(M)依附於一個獨立的硬件線程(Core)。你能夠看到2個Goroutine(G1和G2) 正在並行在各自的OS/硬件線程上面同時執行它們的指令。在每一個邏輯處理器裏，有3個Goroutines以輪轉的方式共享OS線程。這些Goroutines正在以無序的方式併發地執行它們的指令，而且在OS線程上共享時間片。多線程

這裏有一個問題。有些時候利用併發而不採用並行實際上會下降你的吞吐量，有趣的是，有時候利用併發同時加上並行處理也不會爲你帶來你理想中的性能增益。併發

工做負載(workloads)

你是如何知道無序執行(併發)是可行的呢？瞭解你所處理問題的工做負載(workload)是一個起點。有兩種類型的工做負載在併發的時候要考慮到。ide

CPU密集(CPU-Bound)：這種工做負載狀況不會有Goroutines自動切換到waiting狀態的狀況，也不會有自動從waiting狀態切到其餘狀態的狀況。這種狀況發生在進行持續計算的時候。線程計算Pi值就是CPU-Bound。
IO密集(IO-Bound)：這種工做負載會致使Goroutines自動進入等待狀態。這種工做發生在持續地請求網絡資源、或者是進行系統調用、或者是等待事件發生的狀況。一個Goroutines須要讀文件就是IO-Bound。我把同步事件(mutexes，atomic)相似致使Goroutine等待的狀況歸到此類。

cpu-bound的工做負載，你須要並行去使用併發。一個單獨的OS/硬件線程處理多個Goroutines效率很低，由於Goroutines在這個工做負載裏不會主動進入或者是離開等待狀態。Goroutines數多於OS/硬件線程數的時候會下降工做負載的執行速度，由於從OS線程換上或者是換下Goroutines會有延遲(切換的時間)。上下文切換會在workload裏建立出「一切都中止」事件，由於在切換的時候你的全部workload都不會執行。

在IO-Bound的workloads裏，你不須要並行去使用併發。一個單獨OS/硬件線程能夠有效率地處理多個Goroutines，由於Goroutines做爲它本身workload的一部分能夠自動進入或者離開等待狀態(waiting)。Goroutines數量多於OS/硬件線程數能夠加速workload的執行,由於Goroutines在OS線程上切換不會建立「一切都中止」事件。你的workload會天然中止而且這會讓一個不一樣的Goroutine去有效率地使用相同的OS/硬件線程，而不是讓OS/硬件線程空閒下來。

你如何知道每一個硬件線程設置多少個Goroutines會有最好的吞吐量呢？太少的Goroutines你會有更多空閒時間。太多Goroutines你會有更多上下文切換延遲。這件事情你須要考慮，可是這超出了本篇文章講述的範圍。

如今，咱們須要看一些代碼來鞏固你去判斷何時workload能夠利用併發，以及何時須要利用並行何時不須要並行。

整數累加

不須要太複雜的代碼，就看一下下面的add函數。它計算了一堆整數的和。

L1

36 func add(numbers []int) int {
37     var v int
38     for _, n := range numbers {
39         v += n
40     }
41     return v
42 }
複製代碼

在L1的36行，聲明瞭add方法，他接受一個int型的slice，而後返回它們的和。37定義了一個變量v去作數字累加。38行函數遍歷這些整數，39行把當前數加上去。最後41行返回它們的和。

Question: add是否適合無序執行？我相信答案確定是yes。整數集能夠被分解成更小的lists，而且這些lists能夠並行去處理。一旦全部lists都各自加完，這一系列lists的和能夠加到一塊兒，獲得上面代碼裏同樣的結果。

可是，另外一個問題來了。咱們應該分多少個lists去分別單獨處理才能獲得最好的吞吐量呢？爲了回答這個問題，你須要知道add方法運行究竟是哪一種workload。add方法處理的是CPU-Bound類型的workload由於這是一個純數學計算的方法，它不會致使goroutines進入自動等待狀態。這意味着每一個OS/硬件線程一個Goroutine便可得到理想的吞吐量。

下面的L2是add方法的併發版本。

注意：你有多種方式去寫add的併發版本，沒必要去糾結代碼自己。

44 func addConcurrent(goroutines int, numbers []int) int {
45     var v int64
46     totalNumbers := len(numbers)
47     lastGoroutine := goroutines - 1
48     stride := totalNumbers / goroutines
49
50     var wg sync.WaitGroup
51     wg.Add(goroutines)
52
53     for g := 0; g < goroutines; g++ {
54         go func(g int) {
55             start := g * stride
56             end := start + stride
57             if g == lastGoroutine {
58                 end = totalNumbers
59             }
60
61             var lv int
62             for _, n := range numbers[start:end] {
63                 lv += n
64             }
65
66             atomic.AddInt64(&v, int64(lv))
67             wg.Done()
68         }(g)
69     }
70
71     wg.Wait()
72
73     return int(v)
74 }
複製代碼

在L2裏面，addConcurrent方法是add方法的併發版本。這裏有不少代碼所以我只講解重要的代碼行

Line 48：每一個Goroutine會有它單獨的一個小的list去處理。list的size由整數集的size去除以Goroutines的數量獲得。

Line 53：建立goroutines線程池去處理加數操做。

Line 57-59：最後一個goroutines會處理剩下的最後一個list，它可能比其餘list的size要大。

Line 66：全部lists算出來的sum，加到一塊兒獲得最後的一個sum。

併發版本比有序版本更復雜，這種複雜度是否值得呢？回答這個問題最好的方式就是寫一個benchmark。這裏我用了一個一千萬個數大小整數集，而且關掉了垃圾回收。這裏對add和addConcurrent進行了對比。

L3

func BenchmarkSequential(b *testing.B) {
    for i := 0; i < b.N; i++ {
        add(numbers)
    }
}

func BenchmarkConcurrent(b *testing.B) {
    for i := 0; i < b.N; i++ {
        addConcurrent(runtime.NumCPU(), numbers)
    }
}
複製代碼

L3展現了benchmark函數。下面是當Goroutines只有一個單獨的OS/硬件線程能用的狀況。有序版本使用1個Goroutine而後併發版本使用runtime.NumCPU數，個人機器上是8。這個例子下面，併發版本沒有使用並行去作併發。

L4

10 Million Numbers using 8 goroutines with 1 core
2.9 GHz Intel 4 Core i7
Concurrency WITHOUT Parallelism
-----------------------------------------------------------------------------
$ GOGC=off go test -cpu 1 -run none -bench . -benchtime 3s
goos: darwin
goarch: amd64
pkg: github.com/ardanlabs/gotraining/topics/go/testing/benchmarks/cpu-bound
BenchmarkSequential      	    1000	   5720764 ns/op : ~10% Faster
BenchmarkConcurrent      	    1000	   6387344 ns/op
BenchmarkSequentialAgain 	    1000	   5614666 ns/op : ~13% Faster
BenchmarkConcurrentAgain 	    1000	   6482612 ns/op
複製代碼

注意：在你的本機上跑BenchMark很複雜。有不少因素會致使你的benchmarks不夠精確。你的機器儘量的處於空閒狀態這樣能夠去跑一段時間benchmark，以確保本身看到的結果和上面的大致一致。使用測試工具跑兩遍benchmark可以獲得更一致的結果。

L4給出的benchmark代表，在僅有一個單獨OS/硬件線程時候有序版本比並發版本大約要快%10--%13。這在咱們的意料之中，由於併發版本須要在一個單獨的OS線程上頻繁進行上下文切換(context switches)以及處理Goroutines。

下面是每一個Goroutines有一個單獨的OS/硬件線程的狀況下的結果。有序版本用一個Goroutine而後併發版本使用runtime.NumCPU，在我本機上是8個。這種狀況下利用了並行去處理併發。

L5

10 Million Numbers using 8 goroutines with 8 cores
2.9 GHz Intel 4 Core i7
Concurrency WITH Parallelism
-----------------------------------------------------------------------------
$ GOGC=off go test -cpu 8 -run none -bench . -benchtime 3s
goos: darwin
goarch: amd64
pkg: github.com/ardanlabs/gotraining/topics/go/testing/benchmarks/cpu-bound
BenchmarkSequential-8        	    1000	   5910799 ns/op
BenchmarkConcurrent-8        	    2000	   3362643 ns/op : ~43% Faster
BenchmarkSequentialAgain-8   	    1000	   5933444 ns/op
BenchmarkConcurrentAgain-8   	    2000	   3477253 ns/op : ~41% Faster
複製代碼

L5中的benchmark代表了,每一個Goroutines使用一個OS/硬件線程的時候併發版本比有序版本要快大約41%--43%。這是咱們指望中的事情，由於全部的Goroutines如今都在並行執行，8個Goroutines如今都在同一時間併發執行。

排序

須要明白，不是全部的CPU-bound的workloads都適合併發處理。當把工做拆解或者是把結果合併須要花費很大代價的時候這種說法是正確的。這種狀況咱們能夠看一個算法的例子：冒泡排序。看一下下Go實現的冒泡排序。

L6

01 package main
02
03 import "fmt"
04
05 func bubbleSort(numbers []int) {
06     n := len(numbers)
07     for i := 0; i < n; i++ {
08         if !sweep(numbers, i) {
09             return
10         }
11     }
12 }
13
14 func sweep(numbers []int, currentPass int) bool {
15     var idx int
16     idxNext := idx + 1
17     n := len(numbers)
18     var swap bool
19
20     for idxNext < (n - currentPass) {
21         a := numbers[idx]
22         b := numbers[idxNext]
23         if a > b {
24             numbers[idx] = b
25             numbers[idxNext] = a
26             swap = true
27         }
28         idx++
29         idxNext = idx + 1
30     }
31     return swap
32 }
33
34 func main() {
35     org := []int{1, 3, 2, 4, 8, 6, 7, 2, 3, 0}
36     fmt.Println(org)
37
38     bubbleSort(org)
39     fmt.Println(org)
40 }
複製代碼

在L6裏，給出了Go版本的冒泡排序。排序算法遍歷每一個值並在整數集上進行數據交替。根據初始順序不一樣，排序可能須要屢次的遍歷。

Question: bubbleSort的workload適合無序執行嗎？答案確定是no。整數集能夠分解成更小的lists而且這些lists能夠併發地排序。可是全部併發工做完成以後，並無一個有效的方式再去把這些小的lists排序到一塊兒。這裏是一個併發版本的冒泡排序。

L8

01 func bubbleSortConcurrent(goroutines int, numbers []int) {
02     totalNumbers := len(numbers)
03     lastGoroutine := goroutines - 1
04     stride := totalNumbers / goroutines
05
06     var wg sync.WaitGroup
07     wg.Add(goroutines)
08
09     for g := 0; g < goroutines; g++ {
10         go func(g int) {
11             start := g * stride
12             end := start + stride
13             if g == lastGoroutine {
14                 end = totalNumbers
15             }
16
17             bubbleSort(numbers[start:end])
18             wg.Done()
19         }(g)
20     }
21
22     wg.Wait()
23
24     // Ugh, we have to sort the entire list again.
25     bubbleSort(numbers)
26 }
複製代碼

L8中，bubbleSortConcurrent方法是bubbleSort的併發版本。它使用多個Goroutines去併發地排序整個整數集的一部分。結果你獲得的是各自的排序的list。結果你最終在25行仍是要整個list作一次排序。

由於冒泡排序的本質就是遍歷整個list。25行調用bubbleSort直接否認了任何併發的潛在收益。冒泡排序裏，使用併發並無性能上的增益。

讀取文件

咱們給出了2個CPU-Bound類型的workloads，那麼IO-Bound類型的workload狀況是什麼樣的？當Goroutines自動進入或者是離開waiting狀態，狀況會有什麼不一樣麼？看一個IO-bound類型的workload，它的工做內容是讀取文件並查找文本。

第一個版本是一個有序版本的find方法

L10

42 func find(topic string, docs []string) int {
43     var found int
44     for _, doc := range docs {
45         items, err := read(doc)
46         if err != nil {
47             continue
48         }
49         for _, item := range items {
50             if strings.Contains(item.Description, topic) {
51                 found++
52             }
53         }
54     }
55     return found
56 }
複製代碼

在L10裏面，你看到一個有序版本的find函數。line 43定義了一個found變量去存topic在文檔裏的出現次數。line 44，對全部文檔進行遍歷，而且在45行上使用read方法對每一個doc進行讀取。最後從49--53行，使用strings包的Contains方法去檢查topic是否在讀取到的items裏面。若是發現，found變量就對應加一。

這裏是find調用的read方法的實現。

L11

33 func read(doc string) ([]item, error) {
34     time.Sleep(time.Millisecond) // Simulate blocking disk read.
35     var d document
36     if err := xml.Unmarshal([]byte(file), &d); err != nil {
37         return nil, err
38     }
39     return d.Channel.Items, nil
40 }
複製代碼

read方法以一個time.Sleep方法開始。這個裏模擬了真實從硬盤讀取文檔的系統調用所產生的延遲。設置這個延遲對咱們精確地測試有序版本和併發版本find方法的性能差別十分重要。而後在35--39行，測試的xml文檔存儲在fine的全局變量裏，它被反序列化成一個要去處理的struct。最後返回了一個items的集合。

下面是一個併發版本代碼。

注意：有多種方式去寫併發版本代碼，不要糾結於這個代碼自己實現。

L12

58 func findConcurrent(goroutines int, topic string, docs []string) int {
59     var found int64
60
61     ch := make(chan string, len(docs))
62     for _, doc := range docs {
63         ch <- doc
64     }
65     close(ch)
66
67     var wg sync.WaitGroup
68     wg.Add(goroutines)
69
70     for g := 0; g < goroutines; g++ {
71         go func() {
72             var lFound int64
73             for doc := range ch {
74                 items, err := read(doc)
75                 if err != nil {
76                     continue
77                 }
78                 for _, item := range items {
79                     if strings.Contains(item.Description, topic) {
80                         lFound++
81                     }
82                 }
83             }
84             atomic.AddInt64(&found, lFound)
85             wg.Done()
86         }()
87     }
88
89     wg.Wait()
90
91     return int(found)
92 }
複製代碼

L12是find方法的併發版本。併發版本有30行代碼，而非併發版本代碼只有13行。個人目標是處理未知數量的documents時候控制Goroutines的數量。這裏我選擇在池化模式裏使用一個channel去給池子裏的goroutines喂數據。

這部分代碼比較多，我只講解重要部分

Line 61-64: 建立一個channel去處理全部的documents。 Line 65 關閉這個channel，來讓池子裏的goroutines在全部documents處理完成後能自動中止。

Line 70:建立一個goroutines線程池

Line 73--83:每個池子裏的goroutine從channel接受一個document，讀取到內存而後檢查內容是否有topic。匹配的話,lfound就加一個。

Line 84:把每一個單獨goroutines跑出來的數加到一塊兒。

併發版本確實比有序版本代碼更加複雜，這個複雜性是否值得？驗證的最好方式就是再次寫一個benchmark。我用了1000個documents的集合，而且關閉了垃圾回收。一個是順序版本find，一個是併發版本findConcurrent

L13

func BenchmarkSequential(b *testing.B) {
    for i := 0; i < b.N; i++ {
        find("test", docs)
    }
}

func BenchmarkConcurrent(b *testing.B) {
    for i := 0; i < b.N; i++ {
        findConcurrent(runtime.NumCPU(), "test", docs)
    }
}
複製代碼

L13給出了benchmark。下面是當全部goroutines只有一個OS/硬件線程的時候。順序代碼使用1個goroutines，而併發版本是runtime.NumCPU的數，在我本機上是8。這種狀況下，咱們沒用並行去作併發。

L14

10 Thousand Documents using 8 goroutines with 1 core
2.9 GHz Intel 4 Core i7
Concurrency WITHOUT Parallelism
-----------------------------------------------------------------------------
$ GOGC=off go test -cpu 1 -run none -bench . -benchtime 3s
goos: darwin
goarch: amd64
pkg: github.com/ardanlabs/gotraining/topics/go/testing/benchmarks/io-bound
BenchmarkSequential      	       3	1483458120 ns/op
BenchmarkConcurrent      	      20	 188941855 ns/op : ~87% Faster
BenchmarkSequentialAgain 	       2	1502682536 ns/op
BenchmarkConcurrentAgain 	      20	 184037843 ns/op : ~88% Faster
複製代碼

L14裏面代表了，在只有一個單獨OS/硬件線程的時候，併發版本大概要比順序版本代碼快87%--%88。這是咱們預料到的由於每一個Goroutines都能有效的共享這一個OS/硬件線程。在read調用的時候每一個goroutines可以自動進行上下文切換，這樣OS/硬件線程會一直有事情作。

下面是使用並行去作併發處理。

L15

10 Thousand Documents using 8 goroutines with 1 core
2.9 GHz Intel 4 Core i7
Concurrency WITH Parallelism
-----------------------------------------------------------------------------
$ GOGC=off go test -run none -bench . -benchtime 3s
goos: darwin
goarch: amd64
pkg: github.com/ardanlabs/gotraining/topics/go/testing/benchmarks/io-bound
BenchmarkSequential-8        	       3	1490947198 ns/op
BenchmarkConcurrent-8        	      20	 187382200 ns/op : ~88% Faster
BenchmarkSequentialAgain-8   	       3	1416126029 ns/op
BenchmarkConcurrentAgain-8   	      20	 185965460 ns/op : ~87% Faster
複製代碼

L15的benchmark結果說明，額外的OS/硬件線程並無提供更好的性能。