人妖在线一区,国产日韩欧美一区二区综合在线,国产啪精品视频网站免费,欧美内射深插日本少妇

新聞動(dòng)態(tài)

Go并發(fā)編程中sync/errGroup的使用

發(fā)布日期:2022-02-04 08:28 | 文章來源:站長之家

一.序

這一篇算是并發(fā)編程的一個(gè)補(bǔ)充,起因是當(dāng)前有個(gè)項(xiàng)目,大概の 需求是,根據(jù)kafka的分區(qū)(partition)數(shù),創(chuàng)建同等數(shù)量的 消費(fèi)者( goroutine)從不同的分區(qū)中消費(fèi)者消費(fèi)數(shù)據(jù),但是總有某種原因?qū)е?,某一個(gè)分區(qū)消費(fèi)者創(chuàng)建失敗,但是其他分區(qū)消費(fèi)者創(chuàng)建失敗。 最初的邏輯是,忽略分區(qū)失敗的邏輯,將成功創(chuàng)建的分區(qū)消費(fèi)者收集,用于獲取消息進(jìn)行數(shù)據(jù)處理。 代碼就不在這里展示。

問題其實(shí)很明確: 如果在初始化分區(qū)消費(fèi)者時(shí),只要有一個(gè)消費(fèi)創(chuàng)建失敗,那么初始化工作就算失敗,程序應(yīng)該panic,退出。

但是當(dāng)初設(shè)計(jì)時(shí),消費(fèi)者負(fù)責(zé)從kafka上游的某個(gè)topic獲取到數(shù)據(jù),然后經(jīng)過數(shù)據(jù)處理后,再通過生產(chǎn)者將處理后的數(shù)據(jù)發(fā)送到下游的topic中,由于當(dāng)時(shí)設(shè)計(jì)時(shí),代碼耦合比較重,導(dǎo)致無法通過初始化工作做這些,只能在啟動(dòng)生產(chǎn)者后, 再創(chuàng)建消費(fèi)者,這就導(dǎo)致 創(chuàng)建消費(fèi)者-->獲取數(shù)據(jù)-->處理數(shù)據(jù) 雜糅到了一起。 這個(gè)問題一直到最近才有時(shí)間想著來解決。

比如有三個(gè)分區(qū)創(chuàng)建了三個(gè)分區(qū)的消費(fèi)者,每個(gè)分區(qū)的消費(fèi)者對(duì)應(yīng)從自己的分區(qū)重獲取數(shù)據(jù),三個(gè)分區(qū)最初使用waitGroup進(jìn)行控制三個(gè)分區(qū)創(chuàng)建,只有當(dāng)三個(gè)分區(qū)都創(chuàng)建完成后才會(huì)執(zhí)行后續(xù)邏輯。 但是 waitgroup并不能很好的解決:只要一個(gè) goroutine 出錯(cuò)我們就不再等其他 goroutine 了,就默認(rèn)創(chuàng)建分區(qū)消費(fèi)者失敗了,所以此時(shí)便想到了 errGroup

二.errGroup

errGroup 是google開源的基礎(chǔ)擴(kuò)展庫。使用時(shí)先進(jìn)行下載

go get -u golang.org/x/sync

2.1 函數(shù)簽名

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

整個(gè)包就一個(gè) Group 結(jié)構(gòu)體

  • 通過WaitContext 可以創(chuàng)建一個(gè)帶取消的group
  • Go 方法傳入一個(gè) func() error 內(nèi)部會(huì)啟動(dòng)一個(gè)goroutine 去處理
  • Wait 類似WaitGroup的Wait 方法,等待所有的 goroutine結(jié)束后退出,返回的錯(cuò)誤是一個(gè)出錯(cuò)的 err

三.源碼

3.1 Group

type Group struct {
 // context 的 cancel 方法
 cancel func()
 
 // 復(fù)用 WaitGroup
 wg sync.WaitGroup
 
 // 用來保證只會(huì)接受一次錯(cuò)誤
 errOnce sync.Once
 // 保存第一個(gè)返回的錯(cuò)誤
 err  error
}

3.2 WaitContext

func WithContext(ctx context.Context) (*Group, context.Context) {
 ctx, cancel := context.WithCancel(ctx)
 return &Group{cancel: cancel}, ctx
}

WithContext 就是使用 WithCancel 創(chuàng)建一個(gè)可以取消的 context 將 cancel 賦值給 Group 保存起來,然后再將 context 返回回去

注意這里有一個(gè)坑,在后面的代碼中不要把這個(gè) ctx 當(dāng)做父 context 又傳給下游,因?yàn)?errgroup 取消了,這個(gè) context 就沒用了,會(huì)導(dǎo)致下游復(fù)用的時(shí)候出錯(cuò)

3.3 Go

func (g *Group) Go(f func() error) {
 g.wg.Add(1)
 
 go func() {
  defer g.wg.Done()
  
  // 通過執(zhí)行傳入的匿名函數(shù)返回的錯(cuò)誤值判斷是否需要執(zhí)行cancel
  if err := f(); err != nil {
// 這一點(diǎn)很重要,確保錯(cuò)誤只會(huì)被執(zhí)行一次
g.errOnce.Do(func() {
 g.err = err
 if g.cancel != nil {
  g.cancel()
 }
})
  }
 }()
}

Go 方法是一個(gè)封裝,相當(dāng)于go 關(guān)鍵字的加強(qiáng),會(huì)啟動(dòng)一個(gè)攜程,然后利用waitgroup 來控制是否結(jié)束,如果有一個(gè)非 nil 的 error 出現(xiàn)就會(huì)保存起來并且如果有 cancel 就會(huì)調(diào)用 cancel 取消掉,使 ctx 返回

3.4 Wait

func (g *Group) Wait() error {
 g.wg.Wait()
 if g.cancel != nil {
  g.cancel()
 }
 return g.err
}

Wait 方法其實(shí)就是調(diào)用 WaitGroup 等待,如果有 cancel 就調(diào)用一下

四. 案例

基于 errgroup 實(shí)現(xiàn)一個(gè) http server 的啟動(dòng)和關(guān)閉 ,以及 linux signal 信號(hào)的注冊(cè)和處理,要保證能夠 一個(gè)退出,全部注銷退出。

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"))
 })
 
 // 模擬單個(gè)服務(wù)錯(cuò)誤退出
 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 都會(huì)隨之退出
 // 然后 main 函數(shù)中的 g.Wait() 退出,所有協(xié)程都會(huì)退出
 g.Go(func() error {
  err := server.ListenAndServe() // 服務(wù)啟動(dòng)后會(huì)阻塞, 雖然使用的是 go 啟動(dòng),但是由于 g.WaitGroup 試得其是個(gè)阻塞的 協(xié)程
  if err != nil {
log.Println("g1 error,will exit.", err.Error())
  }
  return err
 })
 
 // g2
 // g2 退出了所有的協(xié)程都能退出么?
 // 到調(diào)用 `/shutdown`接口時(shí), serverOut 無緩沖管道寫入數(shù)據(jù), case接收到數(shù)據(jù)后執(zhí)行server.shutdown, 此時(shí) g1 httpServer會(huì)退出
 // g1退出后,會(huì)返回error,將error加到g中,同時(shí)會(huì)調(diào)用 cancel()
 // g3 中會(huì) select case ctx.Done, context 將不再阻塞,g3 會(huì)隨之退出
 // 然后 main 函數(shù)中的 g.Wait() 退出,所有協(xié)程都會(huì)退出
 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)掃描工具會(huì)報(bào)錯(cuò),加上也無傷大雅
  defer cancel()
 
  err := server.Shutdown(timeoutCtx)
  log.Println("shutting down server...")
  return err
 })
 
 // g3
 // g3 捕獲到 os 退出信號(hào)將會(huì)退出
 // g3 退出了所有的協(xié)程都能退出么?
 // g3 退出后, context 將不再阻塞,g2 會(huì)隨之退出
 // g2 退出時(shí),調(diào)用了 shutdown,g1 會(huì)退出
 // 然后 main 函數(shù)中的 g.Wait() 退出,所有協(xié)程都會(huì)退出
 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())
// 當(dāng)g2退出時(shí),已經(jīng)有錯(cuò)誤了,此時(shí)的error 并不會(huì)覆蓋到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())
}

運(yùn)行測(cè)試

D:\gopath\src\Go_base\daily_test\errorGroup>go run demo.go

瀏覽器輸入http://127.0.0.1:8099/shutdown

控制臺(tái)輸出

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é)果可以看出,這種退出可以保證每個(gè)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

分析都在代碼注釋中

五. 參考

https://lailin.xyz/post/go-training-week3-errgroup.html
https://pkg.go.dev/golang.org/x/sync/errgroup

到此這篇關(guān)于Go并發(fā)編程中sync/errGroup的使用的文章就介紹到這了,更多相關(guān)Go sync/errGroup內(nèi)容請(qǐng)搜索本站以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持本站!

香港服務(wù)器租用

版權(quán)聲明:本站文章來源標(biāo)注為YINGSOO的內(nèi)容版權(quán)均為本站所有,歡迎引用、轉(zhuǎn)載,請(qǐng)保持原文完整并注明來源及原文鏈接。禁止復(fù)制或仿造本網(wǎng)站,禁止在非www.sddonglingsh.com所屬的服務(wù)器上建立鏡像,否則將依法追究法律責(zé)任。本站部分內(nèi)容來源于網(wǎng)友推薦、互聯(lián)網(wǎng)收集整理而來,僅供學(xué)習(xí)參考,不代表本站立場(chǎng),如有內(nèi)容涉嫌侵權(quán),請(qǐng)聯(lián)系alex-e#qq.com處理。

相關(guān)文章

實(shí)時(shí)開通

自選配置、實(shí)時(shí)開通

免備案

全球線路精選!

全天候客戶服務(wù)

7x24全年不間斷在線

專屬顧問服務(wù)

1對(duì)1客戶咨詢顧問

在線
客服

在線客服:7*24小時(shí)在線

客服
熱線

400-630-3752
7*24小時(shí)客服服務(wù)熱線

關(guān)注
微信

關(guān)注官方微信
頂部