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

1,235 阅读3分钟

如果让我和别人说说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又是如何实现的呢?核心就在NewChannelStreamPipe这个两个函数里。

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实现,敬请期待……