Golang 标准库 tips之waitgroup详解

来到golang学习网的大家，相信都是编程学习爱好者，希望在这里学习Golang相关编程知识。下面本篇文章就来带大家聊聊《Golang 标准库 tips之waitgroup详解》，介绍一下WaitGroup，希望对大家的知识积累有所帮助，助力实战开发！

WaitGroup 用于线程同步，很多场景下为了提高并发需要开多个协程执行，但是又需要等待多个协程的结果都返回的情况下才进行后续逻辑处理，这种情况下可以通过 WaitGroup 提供的方法阻塞主线程的执行，直到所有的 goroutine 执行完成。
本文目录结构：

WaitGroup 不能被值拷贝
Add 需要在 Wait 之前调用
使用 channel 实现 WaitGroup 的功能
Add 和 Done 数量问题
WaitGroup 和 channel 控制并发数
WaitGroup 和 channel 实现提前退出
WaitGroup 和 channel 返回错误
使用 ErrGroup 返回错误
使用 ErrGroup 实现提前退出
改善版的 Errgroup

WaitGroup 不能被值拷贝

wg 作为一个参数传递的时候，我们在函数中操作的时候还是操作的一个拷贝的变量，对于原来的 wg 是不会改变。
这一点可以从 WaitGroup 实现的源码定义的 struct 能能看出来，WaitGroup 的 struct 就两个字段，第一个字段就是 noCopy，表明这个结构体是不希望直接被复制的。noCopy 是的实现是一个空的 struct{}，主要的作用是嵌入到结构体中作为辅助 vet 工具检查是否通过 copy 赋值这个 WaitGroup 实例，如果有值拷贝的情况，会被检测出来，我们一般的 lint 工具也都能检测出来。
在某些情况下，如果 WaitGroup 需要作为参数传递到其他的方法中，一定需要使用指针类型进行传递。

type WaitGroup struct {
    noCopy noCopy

    // 64-bit value: high 32 bits are counter, low 32 bits are waiter count.
    // 64-bit atomic operations require 64-bit alignment, but 32-bit
    // compilers do not ensure it. So we allocate 12 bytes and then use
    // the aligned 8 bytes in them as state, and the other 4 as storage
    // for the sema.
    state1 [3]uint32
}

可以用以下一个例子来说明：

// 错误的用法，函数传递 wg 是值拷贝
func main() {
    wg := sync.WaitGroup{}

    wg.Add(10)

    for i := 0; i 

<h2>Add 需要在 Wait 之前调用</h2>
<p>WaitGroup 结构体提供了三个方法，Add、Done、Wait，Add 的作用是用来设置WaitGroup的计数值(子goroutine的数量)；Done的作用用来将 WaitGroup 的计数值减 1，其实就是调用Add(-1)；Wait 的作用是检测 WaitGroup 计数器的值是否为 0，如果为 0 表示所有的 goroutine 都运行完成，否则会阻塞等待计数器的值为0(所有的 groutine都执行完成)之后才运行后面的代码。<br>
所以在 WaitGroup 调用的时候一定要保障 Add 函数在 Wait 函数之前执行，否则可能会导致 Wait 方法没有等到所有的结果运行完成而被执行完。也就是我们不能在 Grountine 中来执行 Add 和 Done，这样可能当前 Grountine 来不及运行，外层的 Wait 函数检测到满足条件然后退出了。</p>

<pre class="brush:plain;">
func main() {
    wg := sync.WaitGroup{}
    wg.Wait() // 直接调用 Wait() 方法是不会阻塞的，因为 wg 中 goroutine 计数器的值为 0
    fmt.Println("success")
}

// 错误的写法，在 goroutine 中进行 Add(1) 操作。
// 可能在这些 goroutine 还没来得及 Add(1) 就已经执行 Wait 操作了
func main() {
    wg := sync.WaitGroup{}

    for i := 0; i 

<h2>使用 channel 实现 WaitGroup 的功能</h2>
<p>如果想要实现主线程中等待多个协程的结果都返回的情况下才进行后续调用，也可以通过带缓存区的 channel 来实现，实现的思路是需要先知道等待 groutine 的运行的数量，然后初始化一个相同缓存区数量的 channel，在 groutine 运行结束之后往 channel 中放入一个值，并在主线程中阻塞监听获取 channel 中的值全部返回。</p>

<pre class="brush:plain;">
func main() {
    numGroutine := 10
    ch := make(chan struct{}, numGroutine)

    for i := 0; i 

<h2>Add 和 Done 数量问题</h2>
<p>需要保障 Add 的数量和 Done 的数量一致，如果 Add 数量小于 Done 数量的情况下，调用 Wait 方法会检测到计数器的值为负数，程序会报 panic；如果 Add 数量大于 Done 的数量，会导致 Wait 循环阻塞后面的代码得不到执行。<br>
Add 数量小于 Done 数量：</p>

<pre class="brush:plain;">
func main() {
    wg := sync.WaitGroup{}
    wg.Add(1) // Add 数量小于 Done 数量

    for i := 0; i 

<p>Add 数量大于 Done 数量：</p>

<pre class="brush:plain;">
func main() {
    wg := sync.WaitGroup{}
    wg.Add(20)

    for i := 0; i 

<h2>WaitGroup 和 channel 控制并发数</h2>
<p>使用 waitgroup 可以控制一组 groutine 同时运行并等待结果返回之后再进行后续操作，虽然 groutine 对资源消耗比较小，但是大量的 groutine 并发对系统的压力还是比较大，所以这种情况如果需要控制 waitgroup 中 groutine 并发数量控制，就可以使用缓存的 channel 控制同时并发的 groutine 数量。</p>

<pre class="brush:plain;">
func main() {
    wg := sync.WaitGroup{}
    wg.Add(200)

    ch := make(chan struct{}, 10) // 控制最大并发数是 10
 
    for i := 0; i 

<p>根据使用 channel 实现 WaitGroup 的功能的思路，我们上面的代码也可以通过两个 channel 进行改造来实现。</p>

<pre class="brush:plain;">
func main() {
    numGroutine := 200 // 运行的 groutine 总数量
    numParallel := 10  // 并发的 groutine 数量

    chTotal := make(chan struct{}, numGroutine)
    chParallel := make(chan struct{}, numParallel)

    for i := 0; i 

<h2>WaitGroup 和 channel 实现提前退出</h2>
<p>用 WaitGroup 协调一组并发 goroutine 的做法很常见，但 WaitGroup 本身也有其不足:<br>
WaitGroup 必须要等待控制的一组 goroutine 全部返回结果之后才往下运行，但是有的情况下我们希望能快速失败，也就是这一组 goroutine 中只要有一个失败了，那么就不应该等到所有 goroutine 结束再结束任务，而是提前结束以避免资源浪费，这个时候就可以使用 channel 配合 WaitGroup 实现提前退出的效果。</p>

<pre class="brush:plain;">
func main() {
    wg := sync.WaitGroup{}
    wg.Add(10)

    ch := make(chan struct{}) // 使用一个 channel 传递退出信号

    for i := 0; i 

<h2>WaitGroup 和 channel 返回错误</h2>
<p>WaitGroup 除了不能快速失败之外还有一个问题就是不能在主线程中获取到 groutine 出错时返回的错误，这种情况下就可以用到 channel 进行错误传递，在主线程中获取到错误。</p>

<pre class="brush:plain;">
// 案例一：groutine 中只要有一个失败了则返回 err 并且回到主协程运行后续代码
func main() {
    wg := sync.WaitGroup{}
    wg.Add(10)

    ch := make(chan error) // 使用一个 channel 传递退出信号

    for i := 0; i 

<h2>使用 ErrGroup 返回错误</h2>
<p>正是由于 WaitGroup 有以上说的一些缺点，Go 团队在实验仓库(golang.org/x)增加了 errgroup.Group 的功能，相比 WaitGroup 增加了错误传递、快速失败、超时取消等功能，相对于通过 channel 和 WaitGroup 组合实现这些功能更方便，也更加推荐。<br>
errgroup.Group 结构体也比较简单，在 sync.WaitGroup 的基础之上包装了一个 error 以及一个 cancel 方法，err 的作用是在 goroutine 出错的时候能够返回，cancel 方法的作用是在出错的时候快速失败。<br>
errgroup.Group 对外暴露了3个方法，WithContext、Go、Wait，没有了 Add、Done 方法，其实 Add 和 Done 是在包装在了 errgroup.Group 的 Go 方法里面了，我们执行的时候不需要关心。</p>

<pre class="brush:plain;">
// A Group is a collection of goroutines working on subtasks that are part of
// the same overall task.
//
// A zero Group is valid and does not cancel on error.
type Group struct {
    cancel func()

    wg sync.WaitGroup

    errOnce sync.Once
    err     error
}

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

// Wait blocks until all function calls from the Go method have returned, then
// returns the first non-nil error (if any) from them.
func (g *Group) Wait() error {
    g.wg.Wait()
    if g.cancel != nil {
        g.cancel()
    }
    return g.err
}

// Go calls the given function in a new goroutine.
//
// The first call to return a non-nil error cancels the group; its error will be
// returned by Wait.
func (g *Group) Go(f func() error) {
    g.wg.Add(1)

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

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

以下是使用 errgroup.Group 来实现返回 goroutine 错误的例子：

func main() {
    eg := errgroup.Group{}

    for i := 0; i 

<p>需要注意的一点是通过 errgroup.Group 来返回 err 只会返回其中一个 groutine 的错误，而且是最先返回 err 的 groutine 的错误，这一点是通过 errgroup.Group 的 errOnce 来实现的。</p>
<h2>使用 ErrGroup 实现提前退出</h2>
<p>使用 errgroup.Group 实现提前退出也比较简单，调用 errgroup.WithContext 方法获取 errgroup.Group 对象以及一个可以取消的 WithCancel 的 context，并且将这个 context 方法传入到所有的 groutine 中，并在 groutine 中使用 select 监听这个 context 的 Done() 事件，如果监听到了表明接收到了 cancel 信号，然后退出 groutine 即可。需要注意的是 eg.Go 一定要返回一个 err 才会触发 errgroup.Group 执行 cancel 方法。</p>

<pre class="brush:plain;">
// 案例一：通过 groutine 显示返回 err 触发 errgroup.Group 底层的 cancel 方法
func main() {
    ctx := context.Background()
    eg, ctx := errgroup.WithContext(ctx)

    for i := 0; i 

<h2>改善版的 Errgroup</h2>
<p>使用 errgroup.Group 的 WithContext 我们注意到在返回 eg 对象的同时还会返回另外一个可以取消的 context 对象，这个 context 对象的功能就是用来传递到 eg 需要同步的 groutine 中有一个发生错误时取消整个同步的 groutine，但是有不少同学可能会不经意将这个 context 传到其他的非 eg 同步的业务代码groutine 中，这样会导致非关联的业务代码莫名其妙的收到 cancel 信息，类似如下的写法：</p>

<pre class="brush:plain;">
func main() {
    ctx := context.Background()
    eg, ctx := errgroup.WithContext(ctx)

    for i := 0; i 

<p>另外不管是 WaitGroup 还是 errgroup.Group 都不支持控制最大并发限制以及 panic 恢复的功能，因为我们不能保障我们通过创建的 groutine 不会出现异常，如果没有在创建的协程中捕获异常，会直接导致整个程序退出，这是非常危险的。<br>
这里推荐一下 bilbil 开源的微服务框架 go-kratos/kratos 自己实现了一个改善版本的 errgroup.Group，其实现的的思路是利用 channel 来控制并发，并且创建 errgroup 的时候不会返回 context 避免 context 往非关联的业务方法中传递。</p>
<p>终于介绍完啦！小伙伴们，这篇关于《Golang 标准库 tips之waitgroup详解》的介绍应该让你收获多多了吧！欢迎大家收藏或分享给更多需要学习的朋友吧~golang学习网公众号也会发布Golang相关知识，快来关注吧！</p>