您好,登錄后才能下訂單哦!
本篇內(nèi)容主要講解“Golang中怎么手寫一個簡單的并發(fā)任務manager”,感興趣的朋友不妨來看看。本文介紹的方法操作簡單快捷,實用性強。下面就讓小編來帶大家學習“Golang中怎么手寫一個簡單的并發(fā)任務manager”吧!
ok,我們來試著用 sync 包基礎能力來實現(xiàn)一個簡單的并行任務 manager。首先我們分析下需求。
一定要能做到并發(fā)執(zhí)行各個任務,開多個協(xié)程,而不是在一個 main goroutine 里串行執(zhí)行各個任務;
并發(fā)安全,我們當然不希望出現(xiàn)數(shù)據(jù)異常,不希望并發(fā)執(zhí)行任務導致最后程序因為 runtime error 而掛掉;
如果多個任務都失敗,只返回一個 error 即可;
能夠 recover from panic,不需要開發(fā)者使用的時候再手動去寫 recover 邏輯;
性能有保障。
并發(fā)執(zhí)行這一點我們可以借助 sync.WaitGroup 的能力,每次啟動一個goroutine,WaitGroup 就加 1,在 defer 里完成 Done,啟動所有 goroutine 之后,等著 Wait 返回結(jié)果即可。常規(guī)的能力復用。
需要額外處理的地方在于,怎么實現(xiàn)多個線程只有一個 error 能賦值,以及 recover 的適配。
我們理一下思路,看看代碼怎么寫。
首先一定需要定義一個通用的函數(shù)簽名,使得開發(fā)者能夠傳入自己要執(zhí)行的并發(fā)任務。
type Job interface { Do(ctx context.Context, param interface{}) error Name() string }
我們的 job manager 現(xiàn)階段可以簡單實現(xiàn),只是一組 Job 的集合:
type JobManager []Job
要達到只有一個 error 賦值,且不出現(xiàn) race condition,有兩個方案:
sync.Mutex 加鎖;
sync.Once 只執(zhí)行一次。
當然,什么時候我們都可以用一把大鎖解決問題,但它的性能不會很好,能用原子操作解決的盡量還是不要用 Mutex,這里參照 errgroup,我們可以用一個 Once 對象來控制只賦值一次。
panic 恢復可以直接在 defer 里面 recover 即可,需要能帶出來 stack trace,把它變成一個 error 賦值
有時候我們這個并發(fā)任務數(shù)量非常多,可能還沒創(chuàng)建完 goroutine,某個先創(chuàng)建的任務就已經(jīng)掛了,這時候需要有一個全局的信號,終止后續(xù)的 goroutine 創(chuàng)建。這一點用原子操作就能實現(xiàn)。
把上面的分析落地,這樣我們就實現(xiàn)了一個帶上了 recover 能力,以及終止能力的的 errgroup。
package main import ( "context" "errors" "fmt" "sync" "sync/atomic" ) type Job interface { Do(ctx context.Context, param interface{}) error Name() string } type JobManager []Job func (mgr JobManager) Execute(ctx context.Context, param interface{}) error { var ( stop int32 = 0 err error wg sync.WaitGroup errOnce sync.Once ) for _, job := range mgr { if atomic.LoadInt32(&stop) > 0 { break } wg.Add(1) go func(j Job) { defer func() { wg.Done() if r := recover(); r != nil { errMsg := fmt.Sprintf("JobManager panic: job: %v, reason: %v", j.Name(), r) nerr := errors.New(errMsg) errOnce.Do(func() { if err == nil { err = nerr } }) atomic.AddInt32(&stop, 1) } }() nerr := j.Do(ctx, param) if nerr != nil { atomic.AddInt32(&stop, 1) errOnce.Do(func() { if err == nil { err = nerr } }) } }(job) } wg.Wait() return err }
使用方法也很簡單:
var mgr = JobManager{ AJob, BJob, CJob, // 這里的各個 Job 需要實現(xiàn)一開始我們定義的接口 } err := mgr.Execute(ctx, param)
這里我們需要定義統(tǒng)一的 param interface{},建議是一個接口,各個 Job 執(zhí)行完畢后如果有需要寫入的數(shù)據(jù),可以調(diào)用 param 的 Setter 方法寫入,最后直接拿 param 來做后續(xù)邏輯。
到此,相信大家對“Golang中怎么手寫一個簡單的并發(fā)任務manager”有了更深的了解,不妨來實際操作一番吧!這里是億速云網(wǎng)站,更多相關(guān)內(nèi)容可以進入相關(guān)頻道進行查詢,關(guān)注我們,繼續(xù)學習!
免責聲明:本站發(fā)布的內(nèi)容(圖片、視頻和文字)以原創(chuàng)、轉(zhuǎn)載和分享為主,文章觀點不代表本網(wǎng)站立場,如果涉及侵權(quán)請聯(lián)系站長郵箱:is@yisu.com進行舉報,并提供相關(guān)證據(jù),一經(jīng)查實,將立刻刪除涉嫌侵權(quán)內(nèi)容。