Golang并发模型：轻松入门流水线模型
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个阶段：
- 遍历切片，这是生产者。
- 计算平方值。
- 打印结果，这是消费者。
下面这段代码：
producer()负责生产数据，它会把数据写入通道，并把它写数据的通道返回。square()负责从某个通道读数字，然后计算平方，将结果写入通道，并把它的输出通道返回。main()负责启动producer和square，并且还是消费者，读取suqre的结果，并打印出来。
1package main
2
3import (
4 "fmt"
5)
6
7func producer(nums ...int) <-chan int {
8 out := make(chan int)
9 go func() {
10 defer close(out)
11 for _, n := range nums {
12 out <- n
13 }
14 }()
15 return out
16}
17
18func square(inCh <-chan int) <-chan int {
19 out := make(chan int)
20 go func() {
21 defer close(out)
22 for n := range inCh {
23 out <- n * n
24 }
25 }()
26
27 return out
28}
29
30func main() {
31 in := producer(1, 2, 3, 4)
32 ch := square(in)
33
34 // consumer
35 for ret := range ch {
36 fmt.Printf("%3d", ret)
37 }
38 fmt.Println()
39}结果：
1➜ awesome git:(master) ✗ go run hi.go
2 1 4 9 16这是一种原始的流水线模型，这种原始能让我们掌握流水线的思路。
流水线的特点
- 每个阶段把数据通过channel传递给下一个阶段。
- 每个阶段要创建1个goroutine和1个通道，这个goroutine向里面写数据，函数要返回这个通道。
- 有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.png
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，快去实践一把。
版权申明：内容来源网络，版权归原创者所有。除非无法确认，我们都会标明作者及出处，如有侵权烦请告知，我们会立即删除并表示歉意。谢谢。