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

更新時間：2021年12月12日 16:09:43 作者：深度思維者

本文主要介紹了Go并發(fā)編程中sync/errGroup的使用，文中通過示例代碼介紹的非常詳細，具有一定的參考價值，感興趣的小伙伴們可以參考一下

一.序

這一篇算是并發(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，退出。

但是當初設計時，消費者負責從kafka上游的某個topic獲取到數(shù)據(jù)，然后經(jīng)過數(shù)據(jù)處理后，再通過生產(chǎn)者將處理后的數(shù)據(jù)發(fā)送到下游的topic中，由于當時設計時，代碼耦合比較重，導致無法通過初始化工作做這些，只能在啟動生產(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

errGroup 是google開源的基礎擴展庫。使用時先進行下載

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

整個包就一個 Group 結(jié)構體

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

三.源碼

3.1 Group

type Group struct {
    // context 的 cancel 方法
 cancel func()
 
    // 復用 WaitGroup
 wg sync.WaitGroup
 
 // 用來保證只會接受一次錯誤
 errOnce sync.Once
    // 保存第一個返回的錯誤
 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)建一個可以取消的 context 將 cancel 賦值給 Group 保存起來，然后再將 context 返回回去

注意這里有一個坑，在后面的代碼中不要把這個 ctx 當做父 context 又傳給下游，因為 errgroup 取消了，這個 context 就沒用了，會導致下游復用的時候出錯

3.3 Go

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 關鍵字的加強，會啟動一個攜程，然后利用waitgroup 來控制是否結(jié)束，如果有一個非 nil 的 error 出現(xiàn)就會保存起來并且如果有 cancel 就會調(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 方法其實就是調(diào)用 WaitGroup 等待，如果有 cancel 就調(diào)用一下

四. 案例

基于 errgroup 實現(xiàn)一個 http server 的啟動和關閉，以及 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"))
 })
 
 // 模擬單個服務錯誤退出
 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() // 服務啟動后會阻塞, 雖然使用的是 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

分析都在代碼注釋中