Golang并发模型：流水线模型

原文作者：shitaibin

链接：https://www.jianshu.com/p/8aa03db51043

來源：简书

Golang作为一个实用主义的编程语言，非常注重性能，在语言特性上天然支持并发，它有多种并发模型，通过流水线模型系列文章，你会更好的使用Golang并发特性，提高你的程序性能。

这篇文章主要介绍流水线模型的流水线概念，后面文章介绍流水线模型的FAN-IN和FAN-OUT，最后介绍下如何合理的关闭流水线的协程。

Golang的并发核心思路

Golang并发核心思路是关注数据流动。数据流动的过程交给channel，数据处理的每个环节都交给goroutine，把这些流程画起来，有始有终形成一条线，那就能构成流水线模型。

但我们先从简单的入手。

从一个简单的流水线入手

流水线并不是什么新奇的概念，它能极大的提高生产效率，在当代社会流水线非常普遍，我们用的几乎任何产品（手机、电脑、汽车、水杯），都是从流水线上生产出来的。以汽车为例，整个汽车流水线要经过几百个组装点，而在某个组装点只组装固定的零部件，然后传递给下一个组装点，最终一台完整的汽车从流水线上生产出来。

Golang的并发模型灵感其实都来自我们生活，对软件而言，高的生产效率就是高的性能。

在Golang中，流水线由多个阶段组成，每个阶段之间通过channel连接，每个节点可以由多个同时运行的goroutine组成。

从最简单的流水线入手。下图的流水线由3个阶段组成，分别是A、B、C，A和B之间是通道aCh，B和C之间是通道bCh，A生成数据传递给B，B生成数据传递给C。

流水线中，第一个阶段的协程是生产者，它们只生产数据。最后一个阶段的协程是消费者，它们只消费数据。下图中A是生成者，C是消费者，而B只是中间过程的处理者。

简单流水线.png

举个例子，设计一个程序：计算一个整数切片中元素的平方值并把它打印出来。非并发的方式是使用for遍历整个切片，然后计算平方，打印结果。

我们使用流水线模型实现这个简单的功能，从流水线的角度，可以分为3个阶段：

1. 遍历切片，这是生产者。
2. 计算平方值。
3. 打印结果，这是消费者。

下面这段代码：

• producer()负责生产数据，它会把数据写入通道，并把它写数据的通道返回。
• square()负责从某个通道读数字，然后计算平方，将结果写入通道，并把它的输出通道返回。
• main()负责启动producer和square，并且还是消费者，读取suqre的结果，并打印出来。

package main

import (
    "fmt"
)

func producer(nums ...int) <-chan int {
    out := make(chan int)
    go func() {
        defer close(out)
        for _, n := range nums {
            out <- n
        }
    }()
    return out
}

func square(inCh <-chan int) <-chan int {
    out := make(chan int)
    go func() {
        defer close(out)
        for n := range inCh {
            out <- n * n
        }
    }()

    return out
}

func main() {
    in := producer(1, 2, 3, 4)
    ch := square(in)

    // consumer
    for ret := range ch {
        fmt.Printf("%3d", ret)
    }
    fmt.Println()
}

结果：

1➜  awesome git:(master) ✗ go run hi.go
2  1  4  9 16

这是一种原始的流水线模型，这种原始能让我们掌握流水线的思路。

 

流水线的特点

1. 每个阶段把数据通过channel传递给下一个阶段。
2. 每个阶段要创建1个goroutine和1个通道，这个goroutine向里面写数据，函数要返回这个通道。
3. 有1个函数来组织流水线，我们例子中是main函数。

如果你没了解过流水线，建议自己把以上的程序写一遍，如果遇到问题解决了，那才真正掌握了流水线模型的思路。

接下来，我将介绍流水线模型的FAN-IN、FAN-OUT。

FAN模式可以让我们的流水线模型更好的利用Golang并发，提高软件性能。但FAN模式不一定是万能，不见得能提高程序的性能，甚至还不如普通的流水线。我们先介绍下FAN模式，再看看它怎么提升性能的，它是不是万能的。

 

FAN-IN和FAN-OUT模式

Golang的并发模式灵感来自现实世界，这些模式是通用的，毫无例外，FAN模式也是对当前世界的模仿。以汽车组装为例，汽车生产线上有个阶段是给小汽车装4个轮子，可以把这个阶段任务交给4个人同时去做，这4个人把轮子都装完后，再把汽车移动到生产线下一个阶段。这个过程中，就有任务的分发，和任务结果的收集。其中任务分发是FAN-OUT，任务收集是FAN-IN。

• FAN-OUT模式：多个goroutine从同一个通道读取数据，直到该通道关闭。OUT是一种张开的模式，所以又被称为扇出，可以用来分发任务。
• FAN-IN模式：1个goroutine从多个通道读取数据，直到这些通道关闭。IN是一种收敛的模式，所以又被称为扇入，用来收集处理的结果。

FAN-IN和FAN-OUT实践

我们这次试用FAN-OUT和FAN-IN，解决了前面提到的问题：计算一个整数切片中元素的平方值并把它打印出来。

• producer()保持不变，负责生产数据。
• squre()也不变，负责计算平方值。
• 修改main()，启动3个square，这3个squre从producer生成的通道读数据，这是FAN-OUT。
• 增加merge()，入参是3个square各自写数据的通道，给这3个通道分别启动1个协程，把数据写入到自己创建的通道，并返回该通道，这是FAN-IN。

FAN模式流水线示例：

 1package main
 2
 3import (
 4    "fmt"
 5    "sync"
 6)
 7
 8func producer(nums ...int) <-chan int {
 9    out := make(chan int)
10    go func() {
11        defer close(out)
12        for _, n := range nums {
13            out <- i
14        }
15    }()
16    return out
17}
18
19func square(inCh <-chan int) <-chan int {
20    out := make(chan int)
21    go func() {
22        defer close(out)
23        for n := range inCh {
24            out <- n * n
25        }
26    }()
27
28    return out
29}
30
31func merge(cs ...<-chan int) <-chan int {
32    out := make(chan int)
33
34    var wg sync.WaitGroup
35
36    collect := func(in <-chan int) {
37        defer wg.Done()
38        for n := range in {
39            out <- n
40        }
41    }
42
43    wg.Add(len(cs))
44    // FAN-IN
45    for _, c := range cs {
46        go collect(c)
47    }
48
49    // 错误方式：直接等待是bug，死锁，因为merge写了out，main却没有读
50    // wg.Wait()
51    // close(out)
52
53    // 正确方式
54    go func() {
55        wg.Wait()
56        close(out)
57    }()
58
59    return out
60}
61
62func main() {
63    in := producer(1, 2, 3, 4)
64
65    // FAN-OUT
66    c1 := square(in)
67    c2 := square(in)
68    c3 := square(in)
69
70    // consumer
71    for ret := range merge(c1, c2, c3) {
72        fmt.Printf("%3d ", ret)
73    }
74    fmt.Println()
75}

3个squre协程并发运行，结果顺序是无法确定的，所以你得到的结果，不一定与下面的相同。

1➜  awesome git:(master) ✗ go run hi.go
2  1   4  16   9 

 

FAN模式真能提升性能吗？

相信你心里已经有了答案，可以的。我们还是使用老问题，对比一下简单的流水线和FAN模式的流水线，修改下代码，增加程序的执行时间：

• produer()使用参数生成指定数量的数据。
• square()增加阻塞操作，睡眠1s，模拟阶段的运行时间。
• main()关闭对结果数据的打印，降低结果处理时的IO对FAN模式的对比。

普通流水线：

 1// hi_simple.go
 2
 3package main
 4
 5import (
 6    "fmt"
 7)
 8
 9func producer(n int) <-chan int {
10    out := make(chan int)
11    go func() {
12        defer close(out)
13        for i := 0; i < n; i++ {
14            out <- i
15        }
16    }()
17    return out
18}
19
20func square(inCh <-chan int) <-chan int {
21    out := make(chan int)
22    go func() {
23        defer close(out)
24        for n := range inCh {
25            out <- n * n
26            // simulate
27            time.Sleep(time.Second)
28        }
29    }()
30
31    return out
32}
33
34func main() {
35    in := producer(10)
36    ch := square(in)
37
38    // consumer
39    for _ = range ch {
40    }
41}

使用FAN模式的流水线：

 1// hi_fan.go
 2package main
 3
 4import (
 5    "sync"
 6    "time"
 7)
 8
 9func producer(n int) <-chan int {
10    out := make(chan int)
11    go func() {
12        defer close(out)
13        for i := 0; i < n; i++ {
14            out <- i
15        }
16    }()
17    return out
18}
19
20func square(inCh <-chan int) <-chan int {
21    out := make(chan int)
22    go func() {
23        defer close(out)
24        for n := range inCh {
25            out <- n * n
26            // simulate
27            time.Sleep(time.Second)
28        }
29    }()
30
31    return out
32}
33
34func merge(cs ...<-chan int) <-chan int {
35    out := make(chan int)
36
37    var wg sync.WaitGroup
38
39    collect := func(in <-chan int) {
40        defer wg.Done()
41        for n := range in {
42            out <- n
43        }
44    }
45
46    wg.Add(len(cs))
47    // FAN-IN
48    for _, c := range cs {
49        go collect(c)
50    }
51
52    // 错误方式：直接等待是bug，死锁，因为merge写了out，main却没有读
53    // wg.Wait()
54    // close(out)
55
56    // 正确方式
57    go func() {
58        wg.Wait()
59        close(out)
60    }()
61
62    return out
63}
64
65func main() {
66    in := producer(10)
67
68    // FAN-OUT
69    c1 := square(in)
70    c2 := square(in)
71    c3 := square(in)
72
73    // consumer
74    for _ = range merge(c1, c2, c3) {
75    }
76}

多次测试，每次结果近似，结果如下：

• FAN模式利用了7%的CPU，而普通流水线CPU只使用了3%，FAN模式能够更好的利用CPU，提供更好的并发，提高Golang程序的并发性能。
• FAN模式耗时10s，普通流水线耗时4s。在协程比较费时时，FAN模式可以减少程序运行时间，同样的时间，可以处理更多的数据。

1➜  awesome git:(master) ✗ time go run hi_simple.go
2go run hi_simple.go  0.17s user 0.18s system 3% cpu 10.389 total
3➜  awesome git:(master) ✗ 
4➜  awesome git:(master) ✗ time go run hi_fan.go
5go run hi_fan.go  0.17s user 0.16s system 7% cpu 4.288 total

也可以使用Benchmark进行测试，看2个类型的执行时间，结论相同。为了节约篇幅，这里不再介绍，方法和结果贴在Gist了，想看的朋友瞄一眼，或自己动手搞搞。

FAN模式一定能提升性能吗？

FAN模式可以提高并发的性能，那我们是不是可以都使用FAN模式？

不行的，因为FAN模式不一定能提升性能。

依然使用之前的问题，再次修改下代码，其他不变：

• squre()去掉耗时。
• main()增加producer()的入参，让producer生产10,000,000个数据。

简单版流水线修改代码：

 1// hi_simple.go
 2
 3func square(inCh <-chan int) <-chan int {
 4    out := make(chan int)
 5    go func() {
 6        defer close(out)
 7        for n := range inCh {
 8            out <- n * n
 9        }
10    }()
11
12    return out
13}
14
15func main() {
16    in := producer(10000000)
17    ch := square(in)
18
19    // consumer
20    for _ = range ch {
21    }
22}

FAN模式流水线修改代码：

 1// hi_fan.go
 2package main
 3
 4import (
 5    "sync"
 6)
 7
 8func square(inCh <-chan int) <-chan int {
 9    out := make(chan int)
10    go func() {
11        defer close(out)
12        for n := range inCh {
13            out <- n * n
14        }
15    }()
16
17    return out
18}
19
20func main() {
21    in := producer(10000000)
22
23    // FAN-OUT
24    c1 := square(in)
25    c2 := square(in)
26    c3 := square(in)
27
28    // consumer
29    for _ = range merge(c1, c2, c3) {
30    }
31}

结果，可以跑多次，结果近似：

1➜  awesome git:(master) ✗ time go run hi_simple.go    
2go run hi_simple.go  9.96s user 5.93s system 168% cpu 9.424 total
3➜  awesome git:(master) ✗ time go run hi_fan.go        
4go run hi_fan.go  23.35s user 11.51s system 297% cpu 11.737 total

从这个结果，我们能看到2点。

• FAN模式可以提高CPU利用率。
• FAN模式不一定能提升效率，降低程序运行时间。

 

优化FAN模式

既然FAN模式不一定能提高性能，如何优化？

不同的场景优化不同，要依具体的情况，解决程序的瓶颈。

我们当前程序的瓶颈在FAN-IN，squre函数很快就完成，merge函数它把3个数据写入到1个通道的时候出现了瓶颈，适当使用带缓冲通道可以提高程序性能，

再修改下代码

• merge()中的out修改为：

1out := make(chan int, 100)

结果：

1➜  awesome git:(master) ✗ time go run hi_fan_buffered.go 
2go run hi_fan_buffered.go  19.85s user 8.19s system 323% cpu 8.658 total

使用带缓存通道后，程序的性能有了较大提升，CPU利用率提高到323%，提升了8%，运行时间从11.7降低到8.6，降低了26%。

FAN模式的特点很简单，相信你已经掌握了，如果记不清了看这里，本文所有代码在该Github仓库。

FAN模式很有意思，并且能提高Golang并发的性能，如果想以后运用自如，用到自己的项目中去，还是要写写自己的Demo，快去实践一把。