【導(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 會相對比較合適。
另外它還支持幾個高級用法:
- Groups:允許你定義多個并行的任務(wù),在最后取任務(wù)結(jié)果的時候,可以一起返回;
- Chords:允許你定義一個回調(diào)任務(wù),在 Group 任務(wù)執(zhí)行完畢后執(zhí)行對應(yīng)的回調(diào)任務(wù);
- Chains:允許你定義串行執(zhí)行的任務(wù),任務(wù)將會被串行執(zhí)行;
說了優(yōu)點,再說說它的缺點:
- 任務(wù)監(jiān)控支持不夠,目前只有分布式追蹤 opentracing 的支持,假如我要使用 prometheus,會比較困難,它的自定義錯誤處理過于簡單,連上下文都不給你;
- 傳入的參數(shù)目前只支持非常簡單的參數(shù),不支持 struct、map,還得定義參數(shù)的類型,這樣的方式會將這個庫限制在 Golang 世界中,而無法拓展適用于其它語言;
P.S.
其實對于 Goroutine 的方案,在以下兩種情況下,可以考慮使用:
- 必須同步返回給用戶請求結(jié)果;
- 服務(wù)器資源足夠,僅僅用 Worker pool 就能降低請求的響應(yīng)時長到可接受范圍;
這兩種方案都會返回請求結(jié)果,失敗的情況下靠客戶端重新請求來解決數(shù)據(jù)丟失的問題。
原文標(biāo)題:Golang 中的異步任務(wù)隊列
文章出處:【微信公眾號:Linux愛好者】歡迎添加關(guān)注!文章轉(zhuǎn)載請注明出處。
-
Channel
+關(guān)注
關(guān)注
0文章
31瀏覽量
11749 -
異步
+關(guān)注
關(guān)注
0文章
62瀏覽量
18024 -
Worker
+關(guān)注
關(guān)注
0文章
8瀏覽量
6448
原文標(biāo)題:Golang 中的異步任務(wù)隊列
文章出處:【微信號:LinuxHub,微信公眾號:Linux愛好者】歡迎添加關(guān)注!文章轉(zhuǎn)載請注明出處。
發(fā)布評論請先 登錄
相關(guān)推薦
評論