如果让我和别人说说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实现，敬请期待……
有疑问加站长微信联系（非本文作者）
