您好,登錄后才能下訂單哦!
本篇內(nèi)容介紹了“Go并發(fā)編程中sync/errGroup怎么使用”的有關(guān)知識,在實際案例的操作過程中,不少人都會遇到這樣的困境,接下來就讓小編帶領(lǐng)大家學習一下如何處理這些情況吧!希望大家仔細閱讀,能夠?qū)W有所成!
這一篇算是并發(fā)編程的一個補充,起因是當前有個項目,大概の 需求是,根據(jù)kafka的分區(qū)(partition)數(shù),創(chuàng)建同等數(shù)量的 消費者( goroutine)從不同的分區(qū)中消費者消費數(shù)據(jù),但是總有某種原因?qū)е?,某一個分區(qū)消費者創(chuàng)建失敗,但是其他分區(qū)消費者創(chuàng)建失敗。 最初的邏輯是,忽略分區(qū)失敗的邏輯,將成功創(chuàng)建的分區(qū)消費者收集,用于獲取消息進行數(shù)據(jù)處理。 代碼就不在這里展示。
問題其實很明確: 如果在初始化分區(qū)消費者時,只要有一個消費創(chuàng)建失敗,那么初始化工作就算失敗,程序應該panic,退出。
但是當初設(shè)計時,消費者負責從kafka上游的某個topic獲取到數(shù)據(jù),然后經(jīng)過數(shù)據(jù)處理后,再通過生產(chǎn)者將處理后的數(shù)據(jù)發(fā)送到下游的topic中,由于當時設(shè)計時,代碼耦合比較重,導致無法通過初始化工作做這些,只能在啟動生產(chǎn)者后, 再創(chuàng)建消費者,這就導致 創(chuàng)建消費者-->獲取數(shù)據(jù)-->處理數(shù)據(jù) 雜糅到了一起。 這個問題一直到最近才有時間想著來解決。
比如有三個分區(qū)創(chuàng)建了三個分區(qū)的消費者,每個分區(qū)的消費者對應從自己的分區(qū)重獲取數(shù)據(jù),三個分區(qū)最初使用waitGroup進行控制三個分區(qū)創(chuàng)建,只有當三個分區(qū)都創(chuàng)建完成后才會執(zhí)行后續(xù)邏輯。 但是 waitgroup并不能很好的解決:只要一個 goroutine 出錯我們就不再等其他 goroutine 了,就默認創(chuàng)建分區(qū)消費者失敗了,所以此時便想到了 errGroup
errGroup 是google開源的基礎(chǔ)擴展庫。使用時先進行下載
go get -u golang.org/x/sync
type Group struct { // contains filtered or unexported fields } func WithContext(ctx context.Context) (*Group, context.Context) func (g *Group) Go(f func() error) func (g *Group) Wait() error
整個包就一個 Group 結(jié)構(gòu)體
通過WaitContext 可以創(chuàng)建一個帶取消的group
Go 方法傳入一個 func() error 內(nèi)部會啟動一個goroutine 去處理
Wait 類似WaitGroup的Wait 方法,等待所有的 goroutine結(jié)束后退出,返回的錯誤是一個出錯的 err
type Group struct { // context 的 cancel 方法 cancel func() // 復用 WaitGroup wg sync.WaitGroup // 用來保證只會接受一次錯誤 errOnce sync.Once // 保存第一個返回的錯誤 err error }
func WithContext(ctx context.Context) (*Group, context.Context) { ctx, cancel := context.WithCancel(ctx) return &Group{cancel: cancel}, ctx }
WithContext 就是使用 WithCancel 創(chuàng)建一個可以取消的 context 將 cancel 賦值給 Group 保存起來,然后再將 context 返回回去
注意這里有一個坑,在后面的代碼中不要把這個 ctx 當做父 context 又傳給下游,因為 errgroup 取消了,這個 context 就沒用了,會導致下游復用的時候出錯
func (g *Group) Go(f func() error) { g.wg.Add(1) go func() { defer g.wg.Done() // 通過執(zhí)行傳入的匿名函數(shù)返回的錯誤值判斷是否需要執(zhí)行cancel if err := f(); err != nil { // 這一點很重要,確保錯誤只會被執(zhí)行一次 g.errOnce.Do(func() { g.err = err if g.cancel != nil { g.cancel() } }) } }() }
Go 方法是一個封裝,相當于go 關(guān)鍵字的加強,會啟動一個攜程,然后利用waitgroup 來控制是否結(jié)束,如果有一個非 nil 的 error 出現(xiàn)就會保存起來并且如果有 cancel 就會調(diào)用 cancel 取消掉,使 ctx 返回
func (g *Group) Wait() error { g.wg.Wait() if g.cancel != nil { g.cancel() } return g.err }
Wait 方法其實就是調(diào)用 WaitGroup 等待,如果有 cancel 就調(diào)用一下
基于 errgroup 實現(xiàn)一個 http server 的啟動和關(guān)閉 ,以及 linux signal 信號的注冊和處理,要保證能夠 一個退出,全部注銷退出。
package main import ( "context" "fmt" "log" "net/http" "os" "os/signal" "syscall" "time" "golang.org/x/sync/errgroup" ) func main() { g, ctx := errgroup.WithContext(context.Background()) mux := http.NewServeMux() mux.HandleFunc("/ping", func(w http.ResponseWriter, r *http.Request) { _, _ = w.Write([]byte("pong")) }) // 模擬單個服務(wù)錯誤退出 serverOut := make(chan struct{}) mux.HandleFunc("/shutdown", func(w http.ResponseWriter, r *http.Request) { serverOut <- struct{}{} }) server := http.Server{ Handler: mux, Addr: ":8099", } // g1 // g1 退出了所有的協(xié)程都能退出么? // g1 退出后, context 將不再阻塞,g2, g3 都會隨之退出 // 然后 main 函數(shù)中的 g.Wait() 退出,所有協(xié)程都會退出 g.Go(func() error { err := server.ListenAndServe() // 服務(wù)啟動后會阻塞, 雖然使用的是 go 啟動,但是由于 g.WaitGroup 試得其是個阻塞的 協(xié)程 if err != nil { log.Println("g1 error,will exit.", err.Error()) } return err }) // g2 // g2 退出了所有的協(xié)程都能退出么? // 到調(diào)用 `/shutdown`接口時, serverOut 無緩沖管道寫入數(shù)據(jù), case接收到數(shù)據(jù)后執(zhí)行server.shutdown, 此時 g1 httpServer會退出 // g1退出后,會返回error,將error加到g中,同時會調(diào)用 cancel() // g3 中會 select case ctx.Done, context 將不再阻塞,g3 會隨之退出 // 然后 main 函數(shù)中的 g.Wait() 退出,所有協(xié)程都會退出 g.Go(func() error { select { case <-ctx.Done(): log.Println("g2 errgroup exit...") case <-serverOut: log.Println("g2, request `/shutdown`, server will out...") } timeoutCtx, cancel := context.WithTimeout(context.Background(), 3*time.Second) // 這里不是必須的,但是如果使用 _ 的話靜態(tài)掃描工具會報錯,加上也無傷大雅 defer cancel() err := server.Shutdown(timeoutCtx) log.Println("shutting down server...") return err }) // g3 // g3 捕獲到 os 退出信號將會退出 // g3 退出了所有的協(xié)程都能退出么? // g3 退出后, context 將不再阻塞,g2 會隨之退出 // g2 退出時,調(diào)用了 shutdown,g1 會退出 // 然后 main 函數(shù)中的 g.Wait() 退出,所有協(xié)程都會退出 g.Go(func() error { quit := make(chan os.Signal, 0) signal.Notify(quit, syscall.SIGINT, syscall.SIGTERM) select { case <-ctx.Done(): log.Println("g3, ctx execute cancel...") log.Println("g3 error,", ctx.Err().Error()) // 當g2退出時,已經(jīng)有錯誤了,此時的error 并不會覆蓋到g中 return ctx.Err() case sig := <-quit: return fmt.Errorf("g3 get os signal: %v", sig) } }) // g.Wait 等待所有 go執(zhí)行完畢后執(zhí)行 fmt.Printf("end, errgroup exiting, %+v\n", g.Wait()) }
運行測試
D:\gopath\src\Go_base\daily_test\errorGroup>go run demo.go
瀏覽器輸入http://127.0.0.1:8099/shutdown
控制臺輸出
2021/12/11 10:52:03 g2, request `/shutdown`, server will out...
2021/12/11 10:52:03 g1 error,will exit. http: Server closed
2021/12/11 10:52:03 g3, ctx execute cancel...
2021/12/11 10:52:03 g3 error, context canceled // 間隔了3s
2021/12/11 10:52:06 shutting down server...
end, errgroup exiting, http: Server closed
從執(zhí)行結(jié)果可以看出,這種退出可以保證每個goroutine都能在完成正在執(zhí)行的工作后退出
在terminal 按ctrl + c
輸出
2021/12/11 10:55:51 g2 errgroup exit...
2021/12/11 10:55:51 g1 error,will exit. http: Server closed
2021/12/11 10:55:51 shutting down server...
end, errgroup exiting, g3 get os signal: interrupt
“Go并發(fā)編程中sync/errGroup怎么使用”的內(nèi)容就介紹到這里了,感謝大家的閱讀。如果想了解更多行業(yè)相關(guān)的知識可以關(guān)注億速云網(wǎng)站,小編將為大家輸出更多高質(zhì)量的實用文章!
免責聲明:本站發(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)容。