?1.介紹

在 Go 語(yǔ)言中，我們可以使用 errgroup? 庫(kù)處理 goroutine 中的錯(cuò)誤。

errgroup 庫(kù)最近更新了，新增支持限制并發(fā)數(shù)量的功能。

本文我們介紹 errgroup 庫(kù)的使用方式和實(shí)現(xiàn)原理。

2.使用方式

errgroup 庫(kù)使用非常簡(jiǎn)單，我們通過(guò)三個(gè)簡(jiǎn)單示例代碼，分別介紹三種使用方式。

基礎(chǔ)使用

func main() {
 eg := errgroup.Group{}
 eg.Go(func() error {
  fmt.Println("go1")
  return nil
 })
 eg.Go(func() error {
  fmt.Println("go2")
  err := errors.New("go2 err")
  return err
 })
 err := eg.Wait()
 if err != nil {
  fmt.Println("err =", err)
 }
}

閱讀上面這段代碼，我們使用 errgroup? 庫(kù)的 Go()? 方法啟動(dòng)兩個(gè) goroutine?，分別模擬錯(cuò)誤 goroutine? 和正常 goroutine。

然后，使用 errgroup? 庫(kù)的 Wait()? 方法判斷是否有 goroutine 返回錯(cuò)誤信息。

附加 cancel 功能

func main() {
 eg, ctx := errgroup.WithContext(context.Background())
 eg.Go(func() error {
  time.Sleep(1 * time.Second)
  select {
  case <-ctx.Done():
   fmt.Println("go1 cancel, err = ", ctx.Err())
  default:
   fmt.Println("go1 run")
  }
  return nil
 })
 eg.Go(func() error {
  err := errors.New("go2 err")
  return err
 })
 err := eg.Wait()
 if err != nil {
  fmt.Println("err =", err)
 }
}

閱讀上面這段代碼，我們使用 errgroup? 庫(kù)的 WithContext()? 函數(shù)，可以附加 cancel 功能。

我們?cè)诘谝粋€(gè)使用 Go()? 方法啟動(dòng)的協(xié)程函數(shù)中，使用 select ... case ... default 監(jiān)聽(tīng)其他協(xié)程是否返回錯(cuò)誤并做出相應(yīng)的邏輯處理。

限制并發(fā)數(shù)量

func main() {
 eg := errgroup.Group{}
 eg.SetLimit(2)
 eg.TryGo(func() error {
  fmt.Println("go1 run")
  return nil
 })
 eg.TryGo(func() error {
  err := errors.New("go2 err")
  return err
 })
 eg.TryGo(func() error {
  fmt.Println("go3 run")
  return nil
 })
 err := eg.Wait()
 if err != nil {
  fmt.Println("err =", err)
 }
}

閱讀上面這段代碼，我們使用 errgroup 庫(kù)新增的限制并發(fā)數(shù)量的功能。

首先，使用 SetLimit()? 方法設(shè)置并發(fā)數(shù)量，然后使用 TryGo()? 方法替換 Go() 方法。

3.實(shí)現(xiàn)原理

我們通過(guò)閱讀 errgroup? 庫(kù)的源碼，簡(jiǎn)單介紹 errgroup 的實(shí)現(xiàn)原理。

我們先閱讀 Group 結(jié)構(gòu)體的源碼。

type Group struct {
 cancel func()

 wg sync.WaitGroup

 sem chan token

 errOnce sync.Once
 err     error
}

在源碼中，我們可以發(fā)現(xiàn) Group? 結(jié)構(gòu)體包含的 5 個(gè)字段，其中 sem 字段是最近為了實(shí)現(xiàn)限制并發(fā)數(shù)量功能而新增的。

通過(guò) Group? 結(jié)構(gòu)體的字段，我們可以看出 errgroup? 實(shí)際上是對(duì) sync? 和 context 的封裝。

其中，cancel? 是使用 context? 的 cancel? 方法；wg? 是使用 sync.WairGroup? 的相關(guān)方法；sem? 是通過(guò) channel? 實(shí)現(xiàn)控制并發(fā)數(shù)量；errOnce? 是使用 sync.Once? 的特性，只保存第一個(gè)返回的 goroutine? 錯(cuò)誤；err? 是 goroutine 返回的錯(cuò)誤。

func (g *Group) Go(f func() error) {
 if g.sem != nil {
  g.sem <- token{}
 }

 g.wg.Add(1)
 go func() {
  defer g.done()

  if err := f(); err != nil {
   g.errOnce.Do(func() {
    g.err = err
    if g.cancel != nil {
     g.cancel()
    }
   })
  }
 }()
}

我們閱讀 errgroup? 庫(kù)的 Go()? 方法，首先，通過(guò)判斷 g.sem? 的值是否是 nil?，如果 g.sem? 的值不是 nil?，說(shuō)明已設(shè)置并發(fā)數(shù)量，就通過(guò)向 g.sem? 中發(fā)送一個(gè)空結(jié)構(gòu)體 token{}，來(lái)?yè)屨假Y源。

如果搶到資源，就啟動(dòng)一個(gè) goroutine?，否則，就阻塞，等待其他正在執(zhí)行的 goroutine 釋放一個(gè)資源。

細(xì)心的讀者可能已經(jīng)發(fā)現(xiàn)，Go()? 方法除了開(kāi)頭新增判斷 g.sem? 的值是否為 nil? 的邏輯代碼之外，defer? 也發(fā)生了變化，由之前的直接調(diào)用 sync.WaitGroup? 的 Done()? 方法，改為調(diào)用 errgroup? 庫(kù)新增的 done() 方法。

done() 方法源碼：

func (g *Group) done() {
 if g.sem != nil {
  <-g.sem
 }
 g.wg.Done()
}

通過(guò)閱讀 done()? 方法的源碼，我們可以發(fā)現(xiàn)，在調(diào)用 sync.WaitGroup? 的 Done()? 方法之前，先判斷 g.sem? 的值是否是 nil?，如果不是 nil，則釋放資源。

我們?cè)匍喿x Wait() 方法的源碼：

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

通過(guò)閱讀 Wait()? 方法的源碼，我們可以發(fā)現(xiàn)它實(shí)際上是封裝 sync.WaitGroup? 的 Wait()? 方法，和 context? 包的 cancel?，并且返回所有運(yùn)行的 goroutine 中第一個(gè)返回的錯(cuò)誤。

最后，我們閱讀新增控制并發(fā)數(shù)量的功能 TryGo()? 方法和 SetLimit() 方法的源碼：

func (g *Group) TryGo(f func() error) bool {
 if g.sem != nil {
  select {
  case g.sem <- token{}:
   // Note: this allows barging iff channels in general allow barging.
  default:
   return false
  }
 }

 g.wg.Add(1)
 go func() {
  defer g.done()

  if err := f(); err != nil {
   g.errOnce.Do(func() {
    g.err = err
    if g.cancel != nil {
     g.cancel()
    }
   })
  }
 }()
 return true
}

通過(guò)閱讀 TryGo()? 方法的源碼，我們可以發(fā)現(xiàn)，它和 Go()? 方法的區(qū)別就是在處理 g.sem 的值上，使用的邏輯不同。

TryGo()? 方法在處理 g.sem? 的值時(shí)，使用 select ... case ... default? 語(yǔ)句，先嘗試一次搶占資源，當(dāng)無(wú)法搶到資源時(shí)，不再阻塞，而是直接返回 false，表示執(zhí)行失敗。

SetLimit() 方法的源碼：

func (g *Group) SetLimit(n int) {
 if n < 0 {
  g.sem = nil
  return
 }
 if len(g.sem) != 0 {
  panic(fmt.Errorf("errgroup: modify limit while %v goroutines in the group are still active", len(g.sem)))
 }
 g.sem = make(chan token, n)
}

通過(guò)閱讀 SetLimit()? 方法的源碼，我們可以看出當(dāng)入?yún)?nbsp;n? 的值小于 0? 時(shí)，直接給 g.sem? 賦值為 nil，表示不限制并發(fā)數(shù)量。

在調(diào)用 SetLimit()? 方法時(shí)，g.sem? 必須是一個(gè)空通道，否則程序會(huì) panic。

除去 SetLimit()? 方法的判斷邏輯代碼，實(shí)際上 SetLimit()? 方法就是創(chuàng)建一個(gè)大小為 n? 的有緩沖 channel。

SetLimit()? 和 TryGo() 通常一起使用。

4.總結(jié)

本文我們介紹 Go 方法提供的 errgroup 庫(kù)，該庫(kù)最近新增了控制并發(fā)數(shù)量的功能。

我們先介紹了三種使用方式，然后通過(guò)閱讀源碼，分析其實(shí)現(xiàn)原理。