Scheduling In Go系列文章
本文主要针对Go语言中的调度。
目录
Part I - OS Scheduler
Part II - Go Scheduler
Part III - Concurrency
Part III - Concurrency
这是一个由三部分组成的系列文章的第三篇，这一系列文章将提供对Go中调度器背后的机制和语义的理解。本文的重点是并发性。
Introduction
当我在处理一个问题的时候，一开始我不会考虑是否并发适合这个问题。我会先寻找顺序化的解决方案并且确定这是有效可以工作的。然后在可读性和技术复核的时候，我会开始思考是否并发是合理并且实用的。有些时候并发是非常适合的而其他时候却不太明显。
在这一系列文章的第一篇文章中，我解释了OS调度器的机器和语义，我认为你如果计划要写多线程代码的话这是很重要的。在第二篇文章中，我解释了Go调度器的语义，我相信这对于理解如何在Go中编写并发代码很重要。在这篇文章中，我将开始把OS和Go调度器的机制和语义结合在一起，以便更深入地理解并发是什么和不是什么。
这篇文章的目标是：
- 对于一个工作负荷是否适合并发提供你必须要考虑的指导语义
- 向你展示不同类型的工作负荷如何更改语义，从而更改你将要做出的工程决策。
What is Concurrency
并发意味着“无顺序”执行。有一组指令，它们原本要按照顺序执行，找到一种方式无序执行这组命令仍然产生同样的结果。你面对的问题是，无序执行是增加价值的是必须很明显的。当我说到价值，这指的是这些复杂的花费会增加足够的性能收入。根据你的问题，无序执行有可能是不可能实现的或者甚至是没有意义的。
当然，理解并发与并行是不一样的也是很重要的。并行的意思是同时执行两个或者更多个指令。这个概念和并发是不一样的。只有当你有至少2个可用的OS和硬件线程并且你至少有两个Goroutine时，并行才是可能的，每个指令独立的在OS或者硬件线程上执行。
在图1中，可以看到两个逻辑处理器（P）的关系图，每个处理器的独立操作系统线程（M）连接到机器上的独立硬件线程（Core）。你可以看到两个Goroutine（G1和G2）并行执行，它们同时在各自的OS/硬件线程上执行它们的指令。在每个逻辑处理器中，三个Goroutine轮流共享各自的OS线程。所有这些Goroutine都是并发运行的，它们不按特定顺序执行指令，并在OS线程上共享时间。
这就是问题所在，有时在没有并行的情况下利用并发实际上会降低吞吐量。有趣的是，有时利用并行的并发并不能给你带来比你认为可以实现的更大的性能增益。
Workloads
你如何得知什么时候无序执行是可行的或者有意义的呢？理解你的问题正在处理的工作负荷类型是一个很好的起点。当考虑到并发，有两种工作负荷是需要重点理解的。
- CPU密集型：这是一个永远不会造成Goroutine自然进出等待状态的情况的工作负荷。这是一项不断进行计算的工作。将Pi计算到第n位的线程将是一种CPU密集型工作。
- IO密集型：这是一个导致Goroutine自然进入等待状态的工作负荷。这种类型的工作包括请求通过网络访问资源、对操作系统进行系统调用或等待事件发生。需要读取文件的Goroutine就是一种IO密集型。我认为同步事件（互斥、原子），这种会导致Goroutine等待的事例，也算是这种类型的事件。
在CPU密集型的工作负荷下，你需要使用并行来达到并发。单个操作系统/硬件线程处理多个goroutine的效率不高，因为Goroutine不会作为其工作负荷的一部分进入或退出等待状态。拥有超过OS或硬件线程数目的Goroutine数目会减慢工作负荷的执行，这是由于将Goroutine换上换下OS线程会花费延时。上下文切换会为你的工作负荷创建一个STW(Stop The world)事件，确保在上下文切换期间没有你的工作负荷任务被执行，否则可能会被执行。
在IO密集型的工作负荷下，你不需要使用并行来实现并发。单个OS或者硬件线程可以高效的处理多个Goroutine，因为Goroutine可以作为其工作负荷的一部分进入或退出等待状态。拥有比/硬件线程更多的Goroutine可以加快工作负荷的执行，这是因为将Goroutine换上换下OS线程的延时花费不会造成STW(Stop The world)事件。你的工作负载自然停止，这允许不同的Goroutine有效地利用同一个OS/硬件线程，而不是让OS/硬件线程处于空闲状态。
你知道每个硬件线程多少个Goroutine提供最好的吞吐量么？如果Goroutine数目太少了那么你就有了更多的空闲时间，而如果Goroutine数目太多了，你就会有更多的上下文切换延时。这是你需要考虑的，但超出了这篇文章的范围。
那么现在，复习一些代码来强化你的能力就非常重要了，去判断何时一个工作负荷可以使用并发，何时不能以及何时需要使用并行，何时不需要。
Adding Numbers
我们不需要复杂的代码来可视化和理解这些语义。看看下面名为add的函数，它对一组整数求和。
Listing 1
https://play.golang.org/p/r9LdqUsEzEz
36 func add(numbers []int) int {
37 var v int
38 for _, n := range numbers {
39 v += n
40 }
41 return v
42 }
在Listing 1的36行，定义了一个名为add的函数，这个函数输入一组整数并且返回这组整数的和。从37行开始，它定义了一个整数v来存放这组整数的和。然后在38行这个函数线性遍历整数集合，在39行每个整数都加到当前v上。最后在第41行，这个函数给调用者返回最终的和。
问题：这个add函数适合无序执行么？我相信答案一定是肯定的。这一组整数可以分裂成更小的列表并且这些列表可以并发执行。一旦更小的列表被相加完，这一系列和可以被加在一起最终产生和顺序相加版本一样的结果。
然而，我们也会想到另一个问题。我们应该创建和处理多少个更小的列表从而获得最好的吞吐量呢？为了回答这个问题，你必须知道add使用的是哪种类型的工作负荷。add函数使用的是CPU密集型工作负荷，因为这个算法是单纯使用数学计算的，它不会做任何使它自然进入等待状态的事情。这说明好的吞吐量所需要的就是每个OS或者硬件线程使用一个Goroutine。
下面Listing 2是我并发版本的add。
注意：当你编写并发版本的add时，你可以使用几个其他的方式，采用其他的选择。这时不要纠结于我自己的这种实现方式。如果你有一个更可读的版本，并且它执行起来效果相同甚至更好，我希望你能分享它。
Listing 2
https://play.golang.org/p/r9LdqUsEzEz
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 }
在Listing 2中，addConcurrent函数是add函数的并发版本，与5行的非并发版本代码相比，并发版本使用了26行代码。代码函数比较多，这里我们只关注需要重点理解的行。
48行：每个Goroutine将要将唯一的并且比原数字集合更小的集合相加。这个更小的集合的大小等于总的集合数字数目除以Goroutine的数目。
53行：创建的Goroutine池用来执行加法工作。
57-59行：最后一个Goroutine将添加可能大于其他Goroutine的剩余数字列表。
66行：小列表的总和被合并成一个最终的总和。
并发版本很明显要比串行版本更复杂，但是这个复杂度是不是值得的呢？最好的方式就是创建一个压力测试。在这些压力测试中我使用了一组10 million的数字，并且关闭了垃圾回收。串行版本使用函数add并且并行版本使用函数addConcurrent。
Listing 3
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)
}
}
Listing 3展示了压测函数。这里的结果是对于所有的Goroutine只有一个OS或者硬件线程可以使用。串行版本只使用一个Goroutine，而并行版本使用runtime.NumCPU个Goroutine(对我的电脑来说是8个)。在这个例子中，并发版本使用的是并发而不是并行。
Listing 4
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
注意：在你本地的机器运行一个基准测试是很复杂的。有很多因素会导致你的压测是不正确的。确保你的机器足够空闲并且运行几次基准测试。你要确保你看到结果的一致性。由测试工具运行两次基准测试可以得到最一致的结果。
由Listing 4基准测试结果我们可以看出，对所有Goroutine来说只有一个OS或者硬件线程时，串行版本比并发版本大约快10%~13%。这正是我所期望的，因为并发版本在单个线程上有上下文切换的开销，并且有Goroutine的管理。
下面是当每个Goroutine都有一个单独的OS或者硬件线程可用时的结果。串行版本只使用一个Goroutine，而并行版本使用runtime.NumCPU个Goroutine(对我的电脑来说是8个)。在这种情况下，并发版本利用并行性实现并发性。
Listing 5
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
Listing 5中的基准测试结果显示，当每个Goroutine都有一个单独的OS或者硬件线程可用时，并发版本比串行版本快大约41~43%。这是我所期望的，因为所有Goroutine现在都并行运行，八个Goroutine同时执行它们的并发工作。
Sorting
重要的是要理解，并非所有的CPU密集型工作负荷都适合并发性。这主要是因为打断工作，或者是合并所有结果是代价比较大的。我们可以看到排序算法中的一个例子叫做冒泡排序。看看下面在Go中实现冒泡排序的代码。
*Listing 6
https://play.golang.org/p/S0Us1wYBqG6
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 }
在Listing 6中是一个冒泡排序在Go语言中的实现，这个排序算法扫描一个整数列表，没经过一个值会进行交换值。取决于列表的顺序，在排序完成前可能需要经过多次扫描列表。
问题：冒泡排序函数是适合无序执行的工作负荷吗？我想答案是否定的，整数的集合可以分解成更小的列表，这些列表可以同时排序。但是，在完成所有并发工作之后，没有有效的方法将较小的列表排序在一起。下面是冒泡排序的并发版本的一个例子。
Listing 8
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 }
在Listing 8中，bubbleSortConcurrent函数是bubbleSort函数的一个并发版本。它使用多个Goroutine同时对列表的部分进行排序。但是，剩下的是一个分块排序的值列表。给定一个36个数字的列表，分成12组，如果整个列表没有在第25行再次排序，这将是结果列表。
Listing 9
Before:
25 51 15 57 87 10 10 85 90 32 98 53
91 82 84 97 67 37 71 94 26 2 81 79
66 70 93 86 19 81 52 75 85 10 87 49
After:
10 10 15 25 32 51 53 57 85 87 90 98
2 26 37 67 71 79 81 82 84 91 94 97
10 19 49 52 66 70 75 81 85 86 87 93
由于冒泡排序的本质是遍历列表，所以第25行对bubbleSort函数的调用将否定使用并发性的任何可能的好处。使用冒泡排序时，使用并发不会提高性能。
Reading Files
已经提供了两个CPU密集型的工作负荷，但是IO密集型的工作负荷呢？当Goroutine自然地进入和退出等待状态时，语义是否不同？下面看一下读取文件并执行文本搜索的IO密集型工作负荷。
第一个版本是串行版本，函数是find。
Listing 10
https://play.golang.org/p/8gFe5F8zweN
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 }
在Listing 10中，你可以看到串行版本的find函数。在43行，定义了值value用来存放指定的topic在给定文档中出现的次数。在44行，对文档进行迭代遍历，并且每个文档在45行使用read函数进行读取。最后在49~53行中，使用strings包中的Contains函数来检查给定topic是否在读取到的items中出现。如果找到了topic，则found值增量加一。
下面是由found函数调用的read函数的实现。
Listing 11
https://play.golang.org/p/8gFe5F8zweN
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文档解码为结构体进行处理。最后，在第39行将一组items返回给调用者。
有了串行版本，下面是并发版本。
注意：当你编写并发版本的find时，你可以使用几个其他的方式，采用其他的选择。这时不要纠结于我自己的这种实现方式。如果你有一个更可读的版本，并且它执行起来效果相同甚至更好，我希望你能分享它。
Listing 12
https://play.golang.org/p/8gFe5F8zweN
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 }
在Listing 12中，findConcurrent函数是find函数的并发版本。并发版本使用30行代码，而非并发版本使用13行代码。我实现并发版本的目标是控制用于处理未知数量文档的Goroutine的数量。我选择了一种池模式，使用一个通道来为Goroutine池提供数据。
这里有很多代码，所以我只强调需要理解的重要行。
61~64行：创建一个通道，并填充上所有要处理的文档。
65行：通道已关闭，因此Goroutine池在处理完所有文档时自然终止。
70行：创建Goroutine池。
73~83行：池中的每个Goroutine从通道接收一个文档，将文档读入内存并检查topic是否匹配。当存在匹配项时，局部lFound值变量将递增。
84行：单个Goroutine计数的总和被合并为最终计数。
并发版本肯定比串行版本更复杂，但复杂性是否值得呢？再次回答这个问题的最好方法是创建一个基准测试。对于这些基准测试，我使用了1000个文档的集合，而垃圾回收处于关闭状态。创建了一个使用find函数的串行版本和使用findConcurrent函数的并发版本。
Listing 13
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)
}
}
Listing 13显示了基准测试函数。下面是当所有Goroutine只有一个OS或者硬件线程可用时的结果。串行版本使用1个Goroutine，并发版本在使用runtime.NumCPU个Goroutine，这里在我的机器上是8个Goroutine。在这种情况下，并发版本利用的是无并行性的并发性。
Listing 14
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
Listing 14中的基准测试显示，当只有一个OS或者硬件线程可用于所有Goroutines时，并发版本比串行版本快约87到88%。这是我所期望的，因为所有Goroutine都有效地共享单个OS或者硬件线程。read调用中每个Goroutine的自然的进行上下文切换使得在单个OS或者硬件线程上完成更多的工作。
下面是使用并行性的并发版本的基准测试。
Listing 15
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
Listing 15中的基准测试结果表明，引入额外的OS或者硬件线程并不能提供更好的性能。
Conclusion
这篇文章的目的是在语义上提供一些指导，关于你必须要考虑的东西当你决定一个工作负荷是否适合使用并发性的时候。我试图提供不同类型算法和工作负荷的示例，以便你可以看到语义上的差异以及需要考虑的不同工程决策。
你可以发现，使用IO密集型的工作负荷，不需要并行性就可以大大提高性能。这与你所看到的CPU密集型的工作负荷相反。当涉及到冒泡排序之类的算法时，并发性的使用会增加复杂性，而不会对性能产生任何真正的好处。重要的是确定你的工作负荷是否适合并发性，然后确定必须使用正确语义的工作负荷类型。
本文系翻译，有翻译不当的地方或者其他问题还请多多指教交流~
Scheduling In Go原文索引:
Scheduling In Go : Part I - OS Scheduler
Scheduling In Go : Part II - Go Scheduler
Scheduling In Go : Part III - Concurrency
