Go 语言 Errgroup 库的使用方式和实现原理

时间：2022-10-24 10:20:23人气：次作者：快盘下载我要评论

id="container" data-v-779751fb="" data-element="root">

1.介绍

在 Go 语言中，我们可以使用errgroup库处理goroutine中的错误。

errgroup库最近更新了，新增支持限制并发数量的功能。

本文我们介绍errgroup库的使用方式和实现原理。

2.使用方式

errgroup库使用非常简单，我们通过三个简单示例代码，分别介绍三种使用方式。

基础使用

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库的Go()方法启动两个goroutine，分别模拟错误goroutine和正常goroutine。

然后，使用errgroup库的Wait()方法判断是否有goroutine返回错误信息。

附加 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库的WithContext()函数，可以附加cancel功能。

我们在第一个使用Go()方法启动的协程函数中，使用select ... case ... default监听其他协程是否返回错误并做出相应的逻辑处理。

限制并发数量

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库新增的限制并发数量的功能。

首先，使用SetLimit()方法设置并发数量，然后使用TryGo()方法替换Go()方法。

3.实现原理

我们通过阅读errgroup库的源码，简单介绍errgroup的实现原理。

我们先阅读Group结构体的源码。

type Group struct {
 cancel func()

 wg sync.WaitGroup

 sem chan token

 errOnce sync.Once
 err     error
}

在源码中，我们可以发现Group结构体包含的 5 个字段，其中sem字段是最近为了实现限制并发数量功能而新增的。

通过Group结构体的字段，我们可以看出errgroup实际上是对sync和context的封装。

其中，cancel是使用context的cancel方法；wg是使用sync.WairGroup的相关方法；sem是通过channel实现控制并发数量；errOnce是使用sync.Once的特性，只保存第一个返回的goroutine错误；err是goroutine返回的错误。

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库的Go()方法，首先，通过判断g.sem的值是否是nil，如果g.sem的值不是nil，说明已设置并发数量，就通过向g.sem中发送一个空结构体token{}，来抢占资源。

如果抢到资源，就启动一个goroutine，否则，就阻塞，等待其他正在执行的goroutine释放一个资源。

细心的读者可能已经发现，Go()方法除了开头新增判断g.sem的值是否为nil的逻辑代码之外，defer也发生了变化，由之前的直接调用sync.WaitGroup的Done()方法，改为调用errgroup库新增的done()方法。

done()方法源码：

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

通过阅读done()方法的源码，我们可以发现，在调用sync.WaitGroup的Done()方法之前，先判断g.sem的值是否是nil，如果不是nil，则释放资源。

我们再阅读Wait()方法的源码：

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

通过阅读Wait()方法的源码，我们可以发现它实际上是封装sync.WaitGroup的Wait()方法，和context包的cancel，并且返回所有运行的goroutine中第一个返回的错误。

最后，我们阅读新增控制并发数量的功能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
}

通过阅读TryGo()方法的源码，我们可以发现，它和Go()方法的区别就是在处理g.sem的值上，使用的逻辑不同。

TryGo()方法在处理g.sem的值时，使用select ... case ... default语句，先尝试一次抢占资源，当无法抢到资源时，不再阻塞，而是直接返回false，表示执行失败。

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)
}

通过阅读SetLimit()方法的源码，我们可以看出当入参n的值小于0时，直接给g.sem赋值为nil，表示不限制并发数量。

在调用SetLimit()方法时，g.sem必须是一个空通道，否则程序会panic。

除去SetLimit()方法的判断逻辑代码，实际上SetLimit()方法就是创建一个大小为n的有缓冲channel。

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