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

				}

			})

		}

	}()

}

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

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

3、如果包装了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

}

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

2、如果包装了context的CancelFunc，在出错的时候进行退出操作；

3、抛出第一个出错的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进行超时，取消或者一些异常的情况

巴特西