Arthur's Blog

如果让我和别人说说 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，就是用这个库额外的开销喽。