【译】Golang中的调度（3）：并发编程 - Concurrency

女爷i 2023-06-28 08:41 14阅读 0赞

# 序 #

为了更好理解Go调度器的内在机制，我会以三个部分的内容分别进行阐述，链接如下：

1.  [Golang中的调度（1）：OS调度器 - OS Scheduler][Golang_1_OS_ - OS Scheduler]
2.  [Golang中的调度（2）：Go调度器 - Go Scheduler][Golang_2_Go_ - Go Scheduler]
3.  [Golang中的调度（3）：并发编程- Concurrency][Golang_3_- Concurrency]

本部分内容主要讨论并发编程。

# 引言 #

当我遇到问题，尤其是新问题时，我并不知道是否适合使用并发解决。首先，我会采用顺序化的解决方案并保证它能正常工作，然后，经过易读性和技术审查，我会开始思考并发方式是否合理可行。有时，并发明显很合适，但也有不清楚是否合适的时候。

在第一部分内容中，我解释了OS调度器的机制，如果你计划编写多线程程序，我认为理解它们很重要。在第二部分内容中，我解释了Go调度器的机制，如果你想使用Go编写并发程序，我认为理解它们也很重要。在本部分内容中，我将综合考虑OS和Go调度器的机制，以助你更深入地了解什么是并发。

# 什么是并发 #

并发意味着“乱序”执行，即采取一组原本会按顺序执行的指令，找到一种方法来无序执行它们仍然产生相同的结果。对于这个问题，显而易见的是，无序执行会增加价值。当我说价值时，意思是为复杂性成本增加能获得足够的性能增益。这取决于问题，乱序执行可能无法实现或者毫无意义。

同样重要的是要了解并发与并行并不相同。并行是指同时执行两个或多条指令，这是与并发不同的概念。只有在你拥有至少两个可用的OS/硬件线程并且你至少有两个Goroutines时才可以并行处理，每个Goroutines在每个OS /硬件线程上独立执行指令。

图 1 并发与并行

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NscGhhaGFoYQ_size_16_color_FFFFFF_t_70][]

图1所示，两个逻辑处理器（P1和P2），每个逻辑处理器（P）上的独立OS线程（M）分别连接到计算机上的独立硬件线程（Core），你可以看到两个Goroutine（G1和G2）正在并行执行，同时在各自的OS /硬件线程上执行它们的指令。在每个逻辑处理器中，三个Goroutine轮流共享各自的OS线程。所有这些Goroutine并发运行，以不特定的顺序执行它们的指令，并在OS线程上共享时间。

麻烦的是，有时在没有并行性的情况下利用并发实际上会降低程序吞吐量（throughput）。同样有趣的是，有时将并发与并行结合使用并不会给你带来比原本可以实现的更大的性能提升。

## 任务（Workloads） ##

如何知道何时“乱序”执行是可行的或有意义的？清楚你正在处理的任务类型是一个很好的开始。在考虑并发时，需要了解两种类型的任务。

*  CPU密集型：这种类型的任务永远不会导致Goroutines自然地进入和退出等待态的情况。不断进行计算的任务，如计算Pi到第N位是CPU密集型的工作。
 *  IO密集型：这种类型的任务会导致Goroutines自然进入等待态。这类工作任务包括通过网络访问资源，对操作系统进行系统调用或等待事件发生。一个需要读文件的Goroutine是IO密集型，我将同步事件（互斥量，原子）也归入此类，因为这些事件会导致Goroutine进入等待态。

对于CPU密集型任务，你需要并行地去并发。一个单独的OS /硬件线程处理多个Goroutine效率不高，因为Goroutine不会在执行任务过程中进入和退出等待态。Goroutine的数量多于OS /硬件线程的数量，这会减慢任务的执行速度，这是Goroutine移入和移出OS线程（上下文切换）的延时（Latency）开销所致。Goroutine的上下文切换会为任务造成一个“Stop the world（停滞）”事件，因为在切换期间没有任何任务可以执行。

对于IO密集型任务，你无需并行即可使用并发。单个OS /硬件线程可以高效地处理多个Goroutine，因为Goroutine在执行任务过程中会自然进入和退出等待态。Goroutine的数量多于OS /硬件线程的数量，可以加快任务的执行，因为在OS线程上移入和移出Goroutine的延时开销不会造成“ Stop The World”事件的发生。任务自然暂停，这允许不同的Goroutine有效利用相同的OS /硬件线程，而不是让OS /硬件线程闲置。

如何知道每个硬件线程搭载多少个Goroutine才能提供最佳吞吐量？Goroutine太少，硬件线程将有更多的闲置时间。Goroutine太多，将造成更多的上下文切换延迟开销。这是你要考虑的事情，但超出了本部分内容讨论的范围。

## 加法运算 ##

我们不需要复杂的代码即可可视化并理解这些语义。提供如表1中的add函数，该函数目的是对整数集合求和。

表 1

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

问题：add函数是否适合于无序执行？我相信答案是yes。整数的集合可以分解成更小的集合，并且这些集合可以同时处理。一旦将所有较小的集合求和，就可以将总和加在一起以产生与顺序执行版本相同的答案。

但是，有另外一个问题需要考虑。应该创建和处理几个较小的集合以获得最佳吞吐量呢？要回答这个问题，你必须知道add函数执行的是哪种任务类型。add函数正在执行CPU密集型的任务，因为该函数正在执行纯数学运算，并且没有任何动作会使Goroutine进入自然等待态。这意味着每个OS /硬件线程仅使用一个Goroutine即可获得良好的吞吐量。

下面的表2是并发版的add函数。

注意：如何编写并发版本的add函数，其实可以有很多种方法，所以不要局限在我本次的特定实现上。如果你有可读性更高的版本，且其性能相同或更好，我会很乐意你的分享。

表 2

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 }

在表2中，addConcurrent函数是并发版的add函数。并发版写了26行代码，而非并发版只需要5行。代码行数较多，我只提其中重要的点帮助你理解。

48行：每个Goroutine都会拿到自己单独的但数量较小的数字集合。集合的大小是通过将整个集合的大小除以Goroutine的数量得来的。

53行：创建Goroutine池添加执行任务。

57-59行：最后一个Goroutine将添加可能比其他Goroutine大的剩余数字集合。

66行：将所有小集合的和加在一起成为最终集合总和。

并发版本肯定比顺序版本更复杂，那值得这样做吗？最好的回答方式是使用基准测试（Benchmark）。我使用了1000万个数字的集合，并且关闭了垃圾回收器来进行基准测试，使用add函数与addConcurrent函数进行对比。

表 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)
        }
    }

表3列出了基准测试函数，表4是所有Goroutines只有一个OS /硬件线程可用时的结果，顺序版本使用1个Goroutine，并发版本使用runtime.NumCPU（我机器上8个Goroutine）个Goroutine数来执行。在这种情况下，并发版本没有并行性地并发。

表 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

注意：在本地计算机上运行基准测试是很复杂的，有太多因素会导致基准测试不准确。因此，确保你的计算机尽可能空闲，并多运行几次基准测试。你要确保结果一致，就用测试工具运行两次基准测试以获得最一致的结果。

表4中的基准测试表明，当所有Goroutines仅具有一个OS /硬件线程时，顺序版本比并发版大约快10％到13％。这是我预料之中的，因为并发版本较非并发版，在单个OS线程上会有Goroutines上下文切换的开销。

表 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

如表5所示，每个Goroutine有单独的OS /硬件线程执行。顺序版本使用1个Goroutine，并发版本使用runtime.NumCPU（我机器上8个Goroutine）个Goroutine数来执行。在这种情况下，并发版本并行性地并发。

表5中的基准测试表明，当每个Goroutine都有单独的OS /硬件线程时，并发版本比顺序版本快41％至43％。这符合我的期望，因为所有Goroutine现在都并行运行，八个Goroutine同时执行其并发任务。

## 排序 ##

要知道并非所有CPU密集型的任务都适合并发，当分解工作或合并所有结果的代价非常昂贵时，这就是正确的论点。可以参考被称为冒泡排序的算法示例。以下是Go实现冒泡排序的代码。

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 }

表6是用Go写的冒泡排序的例子。该算法会在扫描集合时进行数字交换，根据集合的顺序，在对所有数字进行排序之前，可能需要对集合进行多次扫描。

问题：bubbleSort函数是否适合乱序执行？我相信答案是no。整数的集合可以分解成更小的集合，并且这些集合可以同时排序。但是，在完成所有并发任务之后，没有有效的方法将较小的集合排序在一起。表7是冒泡排序的并发版本示例。

表 7

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 }

在表7中，给出了bubbleSortConcurrent函数，它是bubbleSort函数的并发版。它使用多个Goroutines同时对集合的各个子集合部分进行排序。如表8所示，给定36个数字，分成若干个子集合，如果所有的子集合结果不能在25行代码时完成有序，那么将再执行一次bubbleSort函数。

表 8

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函数将抵消使用并发所带来的性能收益。因此，冒泡排序，使用并发并不会提高算法性能。

## 读文件 ##

已经提到了两个CPU密集型的任务，那么IO密集型任务呢？当Goroutines自然地进出等待态时，情况是否会有所不同？看一看读取文件并执行文本搜索任务的IO密集型示例。

第一个版本是find函数的顺序版。

表 9

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 }

如表9所示，在第43行，声明了一个名为found的变量，表示在给定文档中找到指定主题的次数。在第44行，遍历文档，并使用read函数在第45行读取文档数据。最后，在第49-53行中，strings.Contains函数用于检查是否可以从文档读取的内容中找到该主题。如果找到主题，则fonud变量加1。

表10是find调用的read函数实现。

表 10

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 }

表10中的read函数以time.Sleep调用开始，睡眠一毫秒用以模拟读文件的阻塞操作。35到39行，解析XML编码的数据并将结果存入d指向的结构体中。最后，在39行将结构体内容返回调用方。

表11是find函数的并发版。

表 11

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 }

如表11所示，并发版本使用了30行代码，而非并发版本只有13行代码。我实现并发版本的目标是控制用于处理未知数量文档的Goroutine数量，因此使用channel在Goroutines池中传递数据。

61到64行，初始化并填充一个传递数据的channel，。

65行，发送方关闭channel，所以当接收方处理完所有文档后，Goroutines都会自然终止。

70行，创建Goroutine池。

73到83行，执行相应子任务，计数lFound。

84行，各个Goroutine计数lFound加在一起成为最终计数found。

并发版本肯定比顺序版本复杂，值得吗？再次回答这个问题的最好方式依然是基准测试。在基准测试中，我使用了1000个文档集，并且关闭了垃圾回收器，比较使用find函数的顺序版和findConcurrent函数的并发版。

表 12

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)
        }
    }

表13 是所有Goroutines只有一个OS /硬件线程可用时的结果，顺序版本使用1个Goroutine，并发版使用runtime.NumCPU（我机器上8个Goroutine）个Goroutine数来执行。在这种情况下，并发版本没有并行性地并发。

表 13

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

表13中的基准测试结果表明，当所有Goroutines仅具有一个OS /硬件线程时，并发版比顺序版快大约87％至88％。这是我预料之中的，因为所有Goroutines都有效地共享了单个OS /硬件线程。Goroutine的上下文切换发生调用的read函数中，这能使得单个OS /硬件线程处理更多的任务。

表14是将并发与并行结合使用时的基准测试结果。

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

表14中的基准测试结果表明，引入额外的OS /硬件线程并没有提供更好的性能。

# 结论 #

这篇文章的目的是提供必须考虑的因素指导，以确定任务是否适合使用并发。我尝试提供不同类型的算法和任务的示例，以便你可以看到因素上的差异以及需要考虑不同的工程决策。

你可以清楚地看到，对于IO密集型任务，不需要并行处理就可以大大提高性能，这与你在CPU密集型任务中看到的相反。对于像冒泡排序这样的算法，并发的使用会增加复杂性，而且也不会带来性能上的增益。确定你要处理的任务是否适合并发，这很重要。

英文原文链接:

[Scheduling In Go : Part III - Concurrency][Scheduling In Go _ Part III - Concurrency]

[Golang_1_OS_ - OS Scheduler]: https://blog.csdn.net/slphahaha/article/details/103499048
[Golang_2_Go_ - Go Scheduler]: https://blog.csdn.net/slphahaha/article/details/103515573
[Golang_3_- Concurrency]: https://blog.csdn.net/slphahaha/article/details/103886865
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NscGhhaGFoYQ_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20200108111054978.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NscGhhaGFoYQ==,size_16,color_FFFFFF,t_70
[Scheduling In Go _ Part III - Concurrency]: https://www.ardanlabs.com/blog/2018/12/scheduling-in-go-part3.html