如何用Golang的channel实现消息的批量处理

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

# 引言 话说,有这样一个场景,就是客户送不断发送消息,需要服务端异步处理。 一个一个的处理未免有些浪费资源,更好的方法是批量处理。 当消息量特别大时,使用kafka之类的message queue自然是首选,但更多的时候,我们想用更加轻量的方案来解决这个问题。 下面来详细分析一下技术需求,这个方案需要实现以下几点: - 消息聚合后处理(最大条数为BatchSize) - 延迟处理(延迟时间为LingerTime) - 自定义错误处理 - 并发处理 # 实现 基于这样的需求,我快速的实现了第一步,消息聚合后处理。 ```go var ( eventQueue = make(chan interface{}, 4) batchSize = 8 workers = 2 batchProcessor = func(messages []interface{}) { fmt.Printf("%+v \n", messages) } ) for i := 0; i < workers; i++ { go func() { var batch []interface{} for { msg := <-eventQueue batch = append(batch, msg) if len(batch) == batchSize { batchProcessor(batch) batch = make([]interface{}, 0) } } }() } for i := 0; i < 100; i++ { eventQueue <- i } ``` 代码虽然简单,但是核心已经有了。 - 带buffer的channel相当于一个FIFO的队列 - 多个常驻的goroutine来提高并发 - goroutine之间是并行的,但每个goroutine内是串行的,所以对batch操作是不用加锁的。 下一步就是添加延迟处理,和错误处理了。 ```go var ( eventQueue = make(chan interface{}, 4) batchSize = 8 workers = 2 lingerTime = 14 * time.Millisecond batchProcessor = func(batch []interface{}) error { fmt.Printf("%+v \n", batch) return nil } errHandler = func(err error, batch []interface{}) { fmt.Println("some error happens") } ) for i := 0; i < workers; i++ { go func() { var batch []interface{} lingerTimer := time.NewTimer(0) if !lingerTimer.Stop() { <-lingerTimer.C } defer lingerTimer.Stop() for { select { case msg := <-eventQueue: batch = append(batch, msg) if len(batch) != batchSize { if len(batch) == 1 { lingerTimer.Reset(lingerTime) } break } if err := batchProcessor(batch); err != nil { errHandler(err, batch) } if !lingerTimer.Stop() { <-lingerTimer.C } batch = make([]interface{}, 0) case <-lingerTimer.C: if err := batchProcessor(batch); err != nil { errHandler(err, batch) } batch = make([]interface{}, 0) } } }() } for i := 0; i < 100; i++ { eventQueue <- i time.Sleep(1 * time.Millisecond) } ``` 虽然只多加了两个点,代码明显复杂了许多,这其实也是很多库的成长过程吧。 一开始专注解决核心问题时,代码还很清晰,当功能逐渐扩展后,代码行数快速增加。 这时,**如果抓不住核心,很容易迷失在代码中**。关于这一点,相信大家在加入一个新的项目,或者看一些成熟项目的源码时都有同感。(*这也是为什么我把不同阶段的代码都列出来的原因,不知各位看官意下如何*) 言归正传,关于代码中为什么使用time.Timer而不是time.After,是因为time.After在for select中使用时,会发生内存泄露。 具体分析,请查看[golang time.After内存泄露问题分析](https://studygolang.com/articles/21870)和[GOLANG中time.After释放的问题](https://studygolang.com/articles/10528)。 所以说呀,**代码写的越多,越容易出bug**,但是功能不完善,代码还是要写的。 实现到这里,当个原型是绰绰有余了,但是要作为一个通用的库,还有很多功能要做,比如说:自定义配置。 最终版的代码,不多不少,正好200行,就不贴过来。有兴趣的同学,请点击[Aggregator.go](https://github.com/Ksloveyuan/channelx/blob/master/aggregator.go)查看。 最后,Aggregator收录在我开源的[channelx](https://github.com/Ksloveyuan/channelx)仓库中,这个库目的是使用channel实现各种好用的轻量级工具。如果有你喜欢用的工具,欢迎点个赞或者star :)

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

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

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