如何把Golang的channel用的如nodejs的stream一样丝滑

如果让我和别人说说Golang有什么特点，我首先想到不一定是goroutine，但一定会是channel。

因为Channel的存在，是让Goroutine威力加成的利器。

如果用一句话来解释channel的作用，我会说

Chanel是一个管道，它会让数据流动起来。

++那么如何理解这个让数据流程起来呢？++

假如说你需要对100次请求，做两种比较耗时的操作，然后再统计加权结果，还需要尽可能的并发来提高性能。示例代码如下：

var multipleChan = make(chan int, 4)
var minusChan = make(chan int, 4)
var harvestChan = make(chan int, 4)

defer close(multipleChan)
defer close(minusChan)
defer close(harvestChan)

go func() {
	for i:=1;i<=100;i++{
		multipleChan <- i
	}
}()

for i:=0; i<4; i++{
	go func() {
		for data := range multipleChan {
			minusChan <- data * 2
			time.Sleep(10* time.Millisecond)
		}
	}()

	go func() {
		for data := range minusChan {
			harvestChan <- data - 1
			time.Sleep(10* time.Millisecond)
		}
	}()
}

var sum = 0
var index = 0
for data := range harvestChan{
	sum += data
	index++
	if index == 100{
		break
	}
}

fmt.Println(sum)

不要笑这段代码简单，如果考虑到错误处理的情况，那还是有些复杂的。比如，某个环节是遇到错误可以忽略，某个环节是遇到要终止所有操作；再加上，有时只关心第一个满足条件的返回值，还需要超时处理。

写一遍也许还可以，要是很多地方都要这样写，那真是头大>_<!!!

重复的代码是万恶之源，Don’t repeat yourself是成为优秀工程师的第一步。

于是，channelx这个库诞生了！

使用了这个库，实现上述同样的功能，代码是这样子的~~

var sum = 0

NewChannelStream(func(seedChan chan<- Result, quitChannel chan struct{}) {
	for i:=1; i<=100;i++{
		seedChan <- Result{Data:i}
	}
	close(seedChan) //记得关闭哦~~~
}).Pipe(func(result Result) Result {
	return Result{Data: result.Data.(int) * 2}
}).Pipe(func(result Result) Result {
	return Result{Data: result.Data.(int) - 1}
}).Harvest(func(result Result) {
	sum += result.Data.(int)
})

fmt.Println(sum)

我喜欢链式风格，所以写成这个样子，你也可以拆开来写的。但重点是代码这样写起来是不是很丝滑，有nodejs stream流的快感呢，嘻嘻~~

除了Pipe->Harvest的组合，还可以实现Pipe->Race, Pipe->Drain, Pipe->Cancel等操作的组合。

这些复杂的例子，都可以参照stream_test.go文件中的单元测试来实现，就不一一贴代码出来了哈。

那么，这个stream又是如何实现的呢？核心就在NewChannelStream和Pipe这个两个函数里。

func NewChannelStream(seedFunc SeedFunc, optionFuncs ...OptionFunc) *ChannelStream {
	cs := &ChannelStream{
		workers:     runtime.NumCPU(),
		optionFuncs: optionFuncs,
	}

	for _, of := range optionFuncs {
		of(cs)
	}

	if cs.quitChan == nil {
		cs.quitChan = make(chan struct{})
	}

	cs.dataChannel = make(chan Item, cs.workers)

	go func() {
		inputChan := make(chan Item)

		go seedFunc(inputChan, cs.quitChan)

	loop:
		for {
			select {
			case <-cs.quitChan:
				break loop

			case res, ok := <-inputChan:
				if !ok {
					break loop
				}

				select {
				case <-cs.quitChan:
					break loop
				default:
				}

				if res.Err != nil {
					cs.errors = append(cs.errors, res.Err)
				}

				if !cs.hasError && res.Err != nil {
					cs.hasError = true
					cs.dataChannel <- res
					if cs.ape == stop {
						cs.Cancel()
					}
					continue
				}

				if cs.hasError && cs.ape == stop {
					continue
				}

				cs.dataChannel <- res
			}
		}

		safeCloseChannel(cs.dataChannel)

	}()

	return cs
}

func (p *ChannelStream) Pipe(dataPipeFunc PipeFunc, optionFuncs ...OptionFunc) *ChannelStream {
	seedFunc := func(dataPipeChannel chan<- Item, quitChannel chan struct{}) {
		wg := &sync.WaitGroup{}
		wg.Add(p.workers)
		for i := 0; i < p.workers; i++ {
			go func() {
				defer wg.Done()
			loop:
				for {
					select {
					case <-quitChannel:
						break loop
					case data, ok := <-p.dataChannel:
						if !ok {
							break loop
						}

						select {
						case <-quitChannel:
							break loop
						default:
						}

						dataPipeChannel <- dataPipeFunc(data)
					}
				}
			}()
		}

		go func() {
			wg.Wait()
			safeCloseChannel(dataPipeChannel)
		}()
	}

	mergeOptionFuncs := make([]OptionFunc, len(p.optionFuncs)+len(optionFuncs)+1)
	copy(mergeOptionFuncs[0:len(p.optionFuncs)], p.optionFuncs)
	copy(mergeOptionFuncs[len(p.optionFuncs):], optionFuncs)
	mergeOptionFuncs[len(p.optionFuncs)+len(optionFuncs)] = passByQuitChan(p.quitChan) //这行保证了整个stream中有一个唯一的quitChan

	return NewChannelStream(seedFunc, mergeOptionFuncs...)
}

代码看着多，刨除初始化的代码、错误处理和退出处理的代码，核心还是通过channel的数据流动。

首先，NewChannelStream中会新建一个inputChan传入seedFunc，然后数据会通过seedChan(即inputChan)，传到dataChannel。

然后，当调用Pipe的时候，Pipe函数会自己创建一个seedFunc从上一个channelStream的dataChannel传到dataPipeChannel中。这个Pipe中的seedFunc又会传入NewChannelStream中，产生一个新channelStream对象，这时在新的channelStream中，inputChan即Pipe中的dataPipeChannel，整个数据流就这样串起来了，过程如下：

inputChan(seedChan)->dataChannel->inputChan(dataPipeChannel)->dataChannel->….

分析过源码，再来看使用ChannelStream的例子和直接用Channel的例子，两个dataChannel分别对应的是multipleChan和minusChan，多出的两个inputChan，就是用这个库额外的开销喽。

原创不易，你的支持就是对我最大的鼓励，欢迎给channelx点个star！:)

未完待续，channelx中还会陆续增加各种常用场景的channel实现，敬请期待……

稀里稀里哗啦啦

Tags