如何用Golang的channel實現消息的批量處理

引言

話說，有這樣一個場景，就是客戶送不斷髮送消息，須要服務端異步處理。

一個一個的處理未免有些浪費資源，更好的方法是批量處理。

當消息量特別大時，使用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 :)