引言
话说，有这样一个场景，就是客户送不断发送消息，需要服务端异步处理。
一个一个的处理未免有些浪费资源，更好的方法是批量处理。
当消息量特别大时，使用kafka之类的message queue自然是首选，但更多的时候，我们想用更加轻量的方案来解决这个问题。
下面来详细分析一下技术需求，这个方案需要实现以下几点：
- 消息聚合后处理（最大条数为BatchSize）
- 延迟处理（延迟时间为LingerTime）
- 自定义错误处理
- 并发处理
实现
基于这样的需求，我快速的实现了第一步，消息聚合后处理。
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操作是不用加锁的。
下一步就是添加延迟处理，和错误处理了。
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内存泄露问题分析和GOLANG中time.After释放的问题。
所以说呀，代码写的越多，越容易出bug，但是功能不完善，代码还是要写的。
实现到这里，当个原型是绰绰有余了，但是要作为一个通用的库，还有很多功能要做，比如说：自定义配置。
最终版的代码，不多不少，正好200行，就不贴过来。有兴趣的同学，请点击Aggregator.go查看。
最后，Aggregator收录在我开源的channelx仓库中，这个库目的是使用channel实现各种好用的轻量级工具。如果有你喜欢用的工具，欢迎点个赞或者star :)
