Go学习之Channel的一些模式

发表于 | 更新于: | 分类于 |

除了在goroutine之间安全的传递数据之外，在看了《Concurrency in Go》之后，感慨channel还有那么多模式可供使用，在个人的学习中总结了以下几种常用的模式

pipeline

概念

我们以爬虫为例，一般爬虫分为如下步骤：

抓取页面 -> 解析页面 -> 整合数据分析 -> 分析结果入库

如果你把上面所有的步骤都放在一个函数里面处理，那会是多难看，多难以维护，从解耦角度考虑，我们可以起四个进程，分别承担不同的角色，例如，进程1负责抓取页面， 进程2负责解析页面，等等，各个进程拿到一个数据后，交给下一个进程来处理，这就是pipeline的基本思想，每个角色只负责关心自己的东西

示例

给定一个数n，执行 (n2 + 1) 2的操作

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
func pipeline() {
	generator := func(done chan interface{}, intergers ...int) <-chan int {
		inStream := make(chan int)
		go func() {
			defer close(inStream)
			for _, i := range intergers {
				select {
				case <-done:
					return
				case inStream <- i:
				}
			}
		}()
		return inStream
	}

	add := func(done <-chan interface{}, inStream <-chan int, increment int) <-chan int {
		addInStream := make(chan int)
		go func() {
			defer close(addInStream)
			for i := range inStream {
				select {
				case <-done:
					return
				case addInStream <- i + increment:
				}
			}
		}()
		return addInStream
	}

	multiply := func(done <-chan interface{}, inStream <-chan int, increment int) <-chan int {
		multiplyInStream := make(chan int)
		go func() {
			defer close(multiplyInStream)
			for i := range inStream {
				select {
				case <-done:
					return
				case multiplyInStream <- i * increment:
				}
			}
		}()
		return multiplyInStream
	}

	done := make(chan interface{})
	defer close(done)
	inStream := generator(done, []int{1, 2, 3, 4, 5, 6, 7}...)
	pipeline := multiply(done, add(done, multiply(done, inStream, 2), 1), 2)

	for v := range pipeline {
		fmt.Println(v)
	}
}

扇入扇出

在pipeline模型中，是一种高效的流式处理，但是假如pipeline中有a,b,c三个环节，b环节处理的特别慢，这时候就会影响到c环节的处理，如果增加b环节进程处理的数量，也就可以减弱b环节的慢处理对整个pipeline的影响，那么a->多个b的过程就是 扇入， 多个b环节输出数据到c环节，就是扇出

示例

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
func FanInFanOut() {
	producer := func(intergers ...int) <-chan interface{} {
		inStream := make(chan interface{})
		go func() {
			defer close(inStream)
			for _, v := range intergers {
				time.Sleep(5 * time.Second)
				inStream <- v
			}
		}()
		return inStream
	}

	fanIn := func(channels ...<-chan interface{},
	) <-chan interface{} {
		var wg sync.WaitGroup
		multiplexStream := make(chan interface{})

		multiplex := func(c <-chan interface{}) {
			defer wg.Done()
			for i := range c {
				multiplexStream <- i
			}
		}

		wg.Add(len(channels))
		for _, c := range channels {
			go multiplex(c)
		}
		go func() {
			wg.Wait()
			close(multiplexStream)
		}()
		return multiplexStream
	}

	consumer := func(inStream <-chan interface{}) {
		for v := range inStream {
			fmt.Println(v)
		}
	}

	nums := runtime.NumCPU()
	producerStreams := make([]<-chan interface{}, nums)
	for i := 0; i < nums; i++ {
		producerStreams[i] = producer(i)
	}

	consumer(fanIn(producerStreams...))
}

tee- channel

概念

假如你从channel中拿到了一条sql语句，这时候，你想对这条sql记录，分析并执行，那你就需要将这条sql分别转发给这三个任务对应的channel，tee-channel 就是做这个事情的

示例

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
func teeChannel() {
	producer := func(intergers ...int) <-chan interface{} {
		inStream := make(chan interface{})
		go func() {
			defer close(inStream)
			for _, v := range intergers {
				inStream <- v
			}
		}()
		return inStream
	}
	tee := func(in <-chan interface{}) (_, _ <-chan interface{}) {
		out1 := make(chan interface{})
		out2 := make(chan interface{})
		go func() {
			defer close(out1)
			defer close(out2)

			for val := range in {
				out1, out2 := out1, out2
				for i := 0; i < 2; i++ {
					select {
					case out1 <- val:
						out1 = nil
					case out2 <- val:
						out2 = nil
					}
				}
			}
		}()
		return out1, out2
	}

	out1, out2 := tee(producer(1, 2, 3, 4, 5))
	for val1 := range out1 {
		fmt.Printf("out1: %v, out2: %v", val1, <-out2)
	}
}

桥接channel

概念

无论是前面提到的pipeline还是扇入扇出，每个goroutine都是对一个channel进行消费，但是实际场景中，可能会有多个channel来供给我们消费，而作为消费者，我们不关心这些值是来自于哪个channel，这种情况下，处理一个充满channel的channel可能会很多。如果我们定义一个功能，可以将充满channel的channel拆解为一个简单的channel，这将使消费者更专注于手头的工作，这就是桥接channel的思想

示例

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
func bridge() {
	gen := func() <-chan <-chan interface{} {
		in := make(chan (<-chan interface{}))
		go func() {
			defer close(in)
			for i := 0; i < 10; i++ {
				stream := make(chan interface{}, 1)
				stream <- i
				close(stream)
				in <- stream
			}
		}()
		return in
	}

	bridge := func(in <-chan (<-chan interface{})) <-chan interface{} {
		valStream := make(chan interface{})
		go func() {
			defer close(valStream)
			for {
				stream := make(<-chan interface{})
				select {
				case maybeStream, ok := <-in:
					if ok == false {
						return
					}
					stream = maybeStream
				}
				for val := range stream {
					valStream <- val
				}
			}
		}()
		return valStream
	}

	for val := range bridge(gen()) {
		fmt.Println(val)
	}
}