理解golang调度之三：并发

前言

这一部分有三篇文章，主要是讲解go调度器的一些内容

三篇文章分别是:

• 理解golang调度之一 ：操做系统调度
• 理解golang调度之二 ：Go调度器
• 理解golang调度之三：并发

简介

当我在解决一个问题尤为是新问题的时候，我开始不会去考虑并发(concurrency)是否合适。我首先会去找一系列的解决方式而后确保它有效。而后在可读性和技术方案评估以后，我会开始去考虑并发是否实际合理。有些时候并发的好处是显而易见的，可是有时候并非很明显。

第一篇文章，我解释了OS调度器的相关内容，我以为这部分对于你写多线程代码很重要。第二篇里，我讲解了一些Go调度器的一些内容，这部分对于你理解和写go的并发代码颇有帮助。在这篇文章里，我会在OS和Go调度器层面让你去深层次的理解并发究竟是什么。

这部份内容的目标是：

• 你的工做负载(workloads)使用并发是否合适，为此提供一些指导建议
• 不一样工做负载的含义，并针对其做出相应的工程方面的决策。

什么是并发

并发的含义就是无序的执行。给你一系列的指令，去找到一个方式能够无序执行并且和有序执行产生一样的结果。这个问题在你面前，显而易见的是无序执行会增长一些足够的性能增益在计算了复杂性成本以后，可是你可能会以为无序执行是不可能的甚至是没有意义的。

你也要清楚一点，并发和并行是不同的。并行是在相同时间内同时执行两个或两个以上的指令，这和并发的概念不同。

图3.1

图3.1里，你看到主机上有两个逻辑处理器。每一个都有他们单独的OS线程(M)依附于一个独立的硬件线程(Core)。你能够看到2个Goroutine(G1和G2) 正在并行在各自的OS/硬件线程上面同时执行它们的指令。在每一个逻辑处理器里，有3个Goroutines以轮转的方式共享OS线程。这些Goroutines正在以无序的方式并发地执行它们的指令，而且在OS线程上共享时间片。

这里有一个问题。有些时候利用并发而不采用并行实际上会下降你的吞吐量，有趣的是，有时候利用并发同时加上并行处理也不会为你带来你理想中的性能增益。

工做负载(workloads)

你是如何知道无序执行(并发)是可行的呢？了解你所处理问题的工做负载(workload)是一个起点。有两种类型的工做负载在并发的时候要考虑到。

• 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/硬件线程并无提供更好的性能。

结论

这篇文章的目的就是让你知道何时你的workload适合使用并发。考虑到不一样的场景，我给出了不一样的例子。

你能够清楚的看到IO-Bound类型的workload并不须要使用并行处理去得到性能的大幅增长，这正好跟CPU-Bound类型的工做截然相反。像相似冒泡算法这种，使用并发其实会增长代码复杂度，并且不会有任何性能增益。因此，必定要肯定你的workload是否适合使用并发场景，这是很重要的事情。

原文连接：www.ardanlabs.com/blog/2018/1…