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

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

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

func (g *Group) Go(f func() error) {
 g.wg.Add(1)
 
 go func() {
  defer g.wg.Done()
        
        // 通過(guò)執(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()
    }
   })
  }
 }()
}

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

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 無(wú)緩沖管道寫(xiě)入數(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ò)，加上也無(wú)傷大雅
  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())
}

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