如何用Worker pool解決異步任務(wù)的問題

【導(dǎo)讀】本文介紹了 Go 移步任務(wù)隊列的實現(xiàn)。

在一些常見的場景中，如果遇到了某些請求特別耗時間，為了不影響其它用戶的請求以及節(jié)約服務(wù)器資源，我們通常會考慮使用異步任務(wù)隊列去解決，這樣可以快速地處理請求、只返回給用戶任務(wù)創(chuàng)建結(jié)果，等待任務(wù)完成之后，我們再告知用戶任務(wù)的完成情況。

對于 Golang，我們可以通過 Worker pool 異步處理任務(wù)，在大多數(shù)情況下，如果不在意數(shù)據(jù)丟失以及服務(wù)器性能足夠，我們就沒有必要考慮別的方案，畢竟這樣實現(xiàn)非常簡單。

接下來我們先來說說如何用 Worker pool 解決異步任務(wù)的問題。

Worker pool

Worker pool，也有叫做 Goroutine pool 的，都是指用 Go channel 以及 Goroutine 來實現(xiàn)的任務(wù)隊列。Go channel 本質(zhì)上就是一個隊列，因此用作任務(wù)隊列是很自然的。

在我們不用 Go channel 的時候，我們也許會使用這樣的方式來處理異步任務(wù)：

fori:=0;i100;i++{
gofunc(){
//processjob
}()
}

這樣的方式是不推薦的，因為在請求量到達(dá)一定程度，系統(tǒng)處理不過來的時候，會造成 Goroutine 的爆炸，拖慢整個系統(tǒng)的運(yùn)行，甚至程序的崩潰，數(shù)據(jù)也就完全丟失了。

如果我們用簡單的方式，可以看看接下來的例子：一個發(fā)送者（也叫做生產(chǎn)者），一個接受者（也叫做消費(fèi)者，或者 Worker）：

typeJobstruct{...}
jobChan:=make(chanJob)
quit:=make(chanbool)
gofunc(){
select{
casejob:=<-jobChan:
???case<-?quit:
???return
}
}()

fori:=0;i100;i++{
jobChan<-?Job{...}
}
close(jobChan)
quit<-?true

如果 Worker 不夠，我們可以增加，這樣可以并行處理任務(wù)：

fori:=0;i10;i++{
gofunc(){
forjob:=rangejobChan{
//processjob
}
}()
}

這樣，一個非常簡單的 Worker pool 就完成了，只是，它對任務(wù)的處理還會有問題，比如無法設(shè)置超時、無法處理 panic 錯誤等。

實際上，目前已經(jīng)有很多的開源庫可以幫你實現(xiàn)了，以worker pool為關(guān)鍵詞在 GitHub 上可以搜到一大堆：

• GitHub - Jeffail/tunny: A goroutine pool for Go
• GitHub - gammazero/workerpool: Concurrency limiting goroutine pool

那么，它們的缺點呢？

很明顯，它們的缺點就在于缺乏管理，可以說是完全不管任務(wù)的結(jié)果，即使我們加日志輸出也只是為了簡單監(jiān)控，更要命的就是進(jìn)程重啟的時候，比如進(jìn)程掛了，或者程序更新，都會導(dǎo)致數(shù)據(jù)丟失，畢竟生產(chǎn)者與消費(fèi)者在一個進(jìn)程中的時候，會互相影響（搶占 CPU 與內(nèi)存資源）。因此前面我也說了，在不管這兩個問題的時候，可以考慮用。

如果數(shù)據(jù)很重要（實際上，我認(rèn)為用戶上傳的業(yè)務(wù)數(shù)據(jù)都重要，不能丟失），為了解決這些問題，我們必須換一種解決方案。

分布式異步任務(wù)隊列

接下來再說說異步的分布式任務(wù)隊列，要用到這個工具的時候，我們大致有以下幾個需求：

• 分布式：生產(chǎn)者與消費(fèi)者隔離；
• 數(shù)據(jù)持久化：在程序重啟的時候，不丟失已有的數(shù)據(jù)；
• 任務(wù)重試：會有任務(wù)偶然失敗的場景，重試是最簡單的方式，但需要保證任務(wù)的執(zhí)行時是冪等的；
• 任務(wù)延時：延遲執(zhí)行，比如 5 分鐘后給用戶發(fā)紅包；
• 任務(wù)結(jié)果的臨時存儲，可用于儲存；
• 任務(wù)處理情況監(jiān)控：及時發(fā)現(xiàn)任務(wù)執(zhí)行出錯情況；

對于 Python 來說，有個大名鼎鼎的 Celery(https://github.com/celery/celery)，它完全包含上面的功能。它包含兩個比較重要的組件：一個是消息隊列，比如 Redis/RabbitMQ 等，Celery中叫做Broker，然后還需要有數(shù)據(jù)庫，用于存儲任務(wù)狀態(tài)，叫做Result Backend。

顯然對于 Go 也有很多不錯的開源庫，其中一個學(xué) Celery 的是 Machinery(github.com/RichardKnop/machinery)，它目前能滿足大部分需求，而且一直在積極維護(hù)，也是我們團(tuán)隊目前在用的。

它目前支持的 Broker 有 AMQP(RabbitMQ)、Redis、AWS SQS、GCP Pub/Sub，目前對國內(nèi)同行來說，RabbitMQ 或者 Redis 會相對比較合適。

另外它還支持幾個高級用法：

1. Groups：允許你定義多個并行的任務(wù)，在最后取任務(wù)結(jié)果的時候，可以一起返回；
2. Chords：允許你定義一個回調(diào)任務(wù)，在 Group 任務(wù)執(zhí)行完畢后執(zhí)行對應(yīng)的回調(diào)任務(wù)；
3. Chains：允許你定義串行執(zhí)行的任務(wù)，任務(wù)將會被串行執(zhí)行；

說了優(yōu)點，再說說它的缺點：

1. 任務(wù)監(jiān)控支持不夠，目前只有分布式追蹤 opentracing 的支持，假如我要使用 prometheus，會比較困難，它的自定義錯誤處理過于簡單，連上下文都不給你；
2. 傳入的參數(shù)目前只支持非常簡單的參數(shù)，不支持 struct、map，還得定義參數(shù)的類型，這樣的方式會將這個庫限制在 Golang 世界中，而無法拓展適用于其它語言；

P.S.

其實對于 Goroutine 的方案，在以下兩種情況下，可以考慮使用：

1. 必須同步返回給用戶請求結(jié)果；
2. 服務(wù)器資源足夠，僅僅用 Worker pool 就能降低請求的響應(yīng)時長到可接受范圍；

這兩種方案都會返回請求結(jié)果，失敗的情況下靠客戶端重新請求來解決數(shù)據(jù)丟失的問題。