go中errgroup源码解读

• errgroup
  • 前言
  • 如何使用
  • 实现原理
  • WithContext
  • Go
  • Wait
  • 错误的使用
  • 总结

errgroup

前言

来看下errgroup的实现

如何使用

func main() {
	var eg errgroup.Group

	eg.Go(func() error {
		return errors.New("test1")
	})

	eg.Go(func() error {
		return errors.New("test2")
	})

	if err := eg.Wait(); err != nil {
		fmt.Println(err)
	}
}

类比于waitgroup,errgroup增长了一个对goroutine错误收集的做用。

不过须要注意的是：

errgroup返回的第一个出错的goroutine抛出的err。

errgroup中还能够加入context

func main() {
	eg, ctx := errgroup.WithContext(context.Background())

	eg.Go(func() error {
		// test1函数还能够在启动不少goroutine
		// 子节点都传入ctx，当test1报错，会把test1的子节点一一cancel
		return test1(ctx)
	})

	eg.Go(func() error {
		return test1(ctx)
	})

	if err := eg.Wait(); err != nil {
		fmt.Println(err)
	}
}

func test1(ctx context.Context) error {
	return errors.New("test2")
}

实现原理

代码很简单

type Group struct {
	// 一个取消的函数，主要来包装context.WithCancel的CancelFunc
	cancel func()

	// 仍是借助于WaitGroup实现的
	wg sync.WaitGroup

	// 使用sync.Once实现只输出第一个err
	errOnce sync.Once

	// 记录下错误的信息
	err     error
}

仍是在WaitGroup的基础上实现的

WithContext

// 返回一个被context.WithCancel从新包装的ctx

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

里面使用了context,经过context.WithCancel对传入的context进行了包装

当WithCancel函数返回的CancelFunc被调用或者是父节点的done channel被关闭（父节点的 CancelFunc 被调用），此 context（子节点）的 done channel 也会被关闭。

errgroup把返回的CancelFunc包进了本身的cancel中，来实现对使用errgroup的ctx启动的goroutine的取消操做。

Go

// 启动取消阻塞的goroutine
// 记录第一个出错的goroutine的err信息
func (g *Group) Go(f func() error) {
	// 借助于waitgroup实现
	g.wg.Add(1)

	go func() {
		defer g.wg.Done()

		// 执行出错
		if err := f(); err != nil {
			// 经过sync.Once记录下第一个出错的err信息
			g.errOnce.Do(func() {
				g.err = err
				// 若是包装了cancel，也就是context的CancelFunc，执行退出操做
				if g.cancel != nil {
					g.cancel()
				}
			})
		}
	}()
}

一、借助于waitgroup实现对goroutine阻塞；

二、经过sync.Once记录下，第一个出错的goroutine的错误信息；

三、若是包装了context的CancelFunc，在出错的时候进行退出操做。

Wait

// 阻塞全部的经过Go加入的goroutine，而后等待他们一个个执行完成
// 而后返回第一个出错的goroutine的错误信息
func (g *Group) Wait() error {
	// 借助于waitgroup实现
	g.wg.Wait()
	// 若是包装了cancel，也就是context的CancelFunc，执行退出操做
	if g.cancel != nil {
		g.cancel()
	}
	return g.err
}

一、借助于waitgroup实现对goroutine阻塞；

二、若是包装了context的CancelFunc，在出错的时候进行退出操做；

三、抛出第一个出错的goroutine的错误信息。

错误的使用

不过工做中发现一个errgroup错误使用的例子

func main() {
	eg := errgroup.Group{}
	var err error
	eg.Go(func() error {
		// 处理业务
		err = test1()
		return err
	})

	eg.Go(func() error {
		// 处理业务
		err = test1()
		return err
	})

	if err = eg.Wait(); err != nil {
		fmt.Println(err)
	}
}

func test1() error {
	return errors.New("test2")
}

很明显err被资源竞争了

$ go run -race main.go 
==================
WARNING: DATA RACE
Write at 0x00c0000801f0 by goroutine 8:
  main.main.func2()
      /Users/yj/Go/src/Go-POINT/sync/errgroup/main.go:23 +0x97
...

总结

errgroup相比比较简单，不过须要先弄明白waitgroup,context以及sync.Once,主要是借助这几个组件来实现的。

errgroup能够带携带context,若是包装了context，会使用context.WithCancel进行超时，取消或者一些异常的状况