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

Ksloveyuan · · 514 次点击 · 开始浏览    置顶
这是一个创建于 的主题,其中的信息可能已经有所发展或是发生改变。

如果让我和别人说说Golang有什么特点,我首先想到不一定是goroutine,但一定会是channel。 因为Channel的存在,是让Goroutine威力加成的利器。 如果用一句话来解释channel的作用,我会说 > Chanel是一个管道,它会让数据流动起来。 ++那么如何理解这个让数据流程起来呢?++ 假如说你需要对100次请求,做两种比较耗时的操作,然后再统计加权结果,还需要尽可能的并发来提高性能。示例代码如下: ```go 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**](https://github.com/Ksloveyuan/channelx)这个库诞生了! 使用了这个库,实现上述同样的功能,代码是这样子的~~ ```go 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](https://github.com/Ksloveyuan/channelx/blob/master/stream_test.go)文件中的单元测试来实现,就不一一贴代码出来了哈。 那么,这个stream又是如何实现的呢?核心就在**NewChannelStream**和**Pipe**这个两个函数里。 ```go 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**](https://github.com/Ksloveyuan/channelx)点个star!:) 未完待续,[**channelx**](https://github.com/Ksloveyuan/channelx)中还会陆续增加各种常用场景的channel实现,敬请期待……

有疑问加站长微信联系(非本文作者)

入群交流(和以上内容无关):加入Go大咖交流群,或添加微信:liuxiaoyan-s 备注:入群;或加QQ群:701969077

514 次点击  
加入收藏 微博
暂无回复
添加一条新回复 (您需要 登录 后才能回复 没有账号 ?)
  • 请尽量让自己的回复能够对别人有帮助
  • 支持 Markdown 格式, **粗体**、~~删除线~~、`单行代码`
  • 支持 @ 本站用户;支持表情(输入 : 提示),见 Emoji cheat sheet
  • 图片支持拖拽、截图粘贴等方式上传