同步协程的必备工具: WaitGroup

1. 简介

本文将介绍 Go 语言中的 WaitGroup 并发原语，包括 WaitGroup 的基本使用方法、实现原理、使用注意事项以及常见的使用方式。能够更好地理解和应用 WaitGroup 来协调多个 Goroutine 的执行，提高 Go 并发编程的效率和稳定性。
2. 基本使用

2.1 定义

WaitGroup是Go语言标准库中的一个结构体，它提供了一种简单的机制，用于同步多个协程的执行。适用于需要并发执行多个任务并等待它们全部完成后才能继续执行后续操作的场景。
2.2 使用方式

首先主协程创建WaitGroup实例，然后在每个协程的开始处，调用Add(1)方法，表示需要等待一个任务执行完成，然后协程在任务执行完成之后，调用Done方法，表示任务已经执行完成了。
主协程中，需要调用Wait()方法，等待所有协程完成任务，示例如下:

1. func main(){
2.     //首先主协程创建WaitGroup实例
3.     var wg sync.WaitGroup
4.     // 开始时调用Add方法表示有个任务开始执行
5.     wg.Add(1)
6.     go func() {
7.         // 开始执行...
8.         //完成之后,调用Done方法
9.         wg.Done()
10.     }()
11.     // 调用Wait()方法，等待所有协程完成任务
12.     wg.Wait()
13.     // 执行后续逻辑
14. }

复制代码

2.3 使用例子

1. package main
3. import (
4.     "fmt"
5.     "sync"
6. )
8. func main() {
9.     var wg sync.WaitGroup
10.     for i := 0; i < 5; i++ {
11.        wg.Add(1)
12.        go func(i int) {
13.           defer wg.Done()
14.           fmt.Printf("任务%d开始执行\n", i)
15.           // 模拟协程任务执行一段时间
16.           time.Sleep(time.Duration(rand.Int() % 100))
17.           // 线程任务执行完成
18.           fmt.Printf("任务%d执行完毕\n", i)
19.        }(i)
20.     }
21.     fmt.Println("主协程开始等待所有任务执行完成...")
22.     wg.Wait()
23.     fmt.Println("所有协程已经执行完毕...")
24. }

复制代码

在这个例子中，我们使用了sync.WaitGroup来等待5个协程执行完毕。在循环中，每创建一个任务，我们调用一次wg.Add(1)方法，然后启动一个协程去执行任务，当协程完成任务后，调用wg.Done方法，告知主协程任务已经执行完毕。然后主协程会在5个协程任务全部执行完毕之后，才会继续向下执行。
3.实现原理

3.1 设计初衷

WaitGroup的设计初衷就是为了等待一组操作完成后再执行下一步操作，通常会在一组协程中使用。
3.2 实现

sync.WaitGroup 结构体中的 state1 和 state2 字段是用于实现 WaitGroup 功能的重要变量。

1. type WaitGroup struct {
2.    noCopy noCopy
4.    state1 uint64
5.    state2 uint32
6. }

复制代码

由于 WaitGroup 需要等待一组操作完成之后再执行，因此需要等待所有操作完成之后才能继续执行。为了实现这个功能，WaitGroup 使用了一个计数器 counter 来记录还有多少个操作没有完成，如果 counter 的值为 0，则表示所有操作已经完成。
同时，WaitGroup 在所有任务都完成之后，需要唤醒所有处于等待的协程，此时需要知道有多少个协程处于等待状态。为了实现这个功能，WaitGroup 使用了一个等待计数器 waiter 来记录当前有多少个协程正在等待操作完成。
这里WaitGroup对于计数器和等待计数器的实现，是通过一个64位无符号整数来实现的，也就是WaitGroup结构体中的state1，其中高32位保存了任务计数器counter的值，低32位保存了等待计数器waiter的值。当我们创建一个 WaitGroup 实例时，该实例的任务计数器和等待计数器都被初始化为 0。
而且，等待协程需要等待所有任务完成之后才能继续执行，所以等待协程在任务未完成时会被阻塞，当任务全部完成后，自动被唤醒。WaitGroup使用 state2 用于实现信号量机制。通过调用 runtime_Semacquire() 和 runtime_Semrelease() 函数，可以在不阻塞线程的情况下进行等待和通知操作。下面是Add，Done和Wait方法的具体实现:
调用 Add() 方法增加/减小counter的值，delta的值可以是正数，也可以是负数，下面是Add方法的源码实现:

1. func (wg *WaitGroup) Add(delta int) {
2.    // delta 的值可以为负数,Done方法便是通过Add(-1)来实现的
3.    // statep: 为state1的地址  semap: 为state2的地址
4.    statep, semap := wg.state()
5.    // 高32位的值 加上 delta,增加任务计数器的值
6.    state := atomic.AddUint64(statep, uint64(delta)<<32)
7.    // v: 取高32位数据,获取到待完成任务数
8.    v := int32(state >> 32)
9.    // 取低32位数据,获取到等待线程的值
10.    w := uint32(state)
11.    // v > 0: 说明还有待完成的任务数,此时不应该唤醒等待协程
12.    // w = 0: 说明没有协程在等待,此时可以直接退出
13.    if v > 0 || w == 0 {
14.       return
15.    }     
16.    // 此时v = 0,所有任务都完成了,唤醒等待协程
17.    *statep = 0
18.    for ; w != 0; w-- {
19.       runtime_Semrelease(semap, false, 0)
20.    }
21. }

复制代码

调用 Done() 方法表示完成了一个任务，通过调用Add方法，delta值为-1，减少任务计数器counter的值，当其归为0时，便自动唤醒所有处于等待的协程。

1. // Done decrements the WaitGroup counter by one.
2. func (wg *WaitGroup) Done() {
3.    wg.Add(-1)
4. }

复制代码

调用Wait方法，等待任务执行完成，增加等待计数器Waiter的值:

1. func (wg *WaitGroup) Wait() {
2.    // statep: 为state1的地址  semap: 为state2的地址
3.    statep, semap := wg.state()
4.    for {
5.       // 加载state1的值
6.       state := atomic.LoadUint64(statep)
7.       // v: 取高32位数据,获取到待完成任务数
8.       v := int32(state >> 32)
9.       // 没有任务待执行,全部都完成了
10.       if v == 0 {
11.          return
12.       }
13.       // 增加waiter计数器的值
14.       if atomic.CompareAndSwapUint64(statep, state, state+1) {
15.           // 等待被唤醒
16.          runtime_Semacquire(semap)
17.          return
18.       }
19.    }
20. }

复制代码

3.3 实现补充

Add方法，Done方法以及Wait方法实现中，有一些异常场景的验证逻辑被我删除掉了。当出现异常场景时，说明用户使用方式和WaitGroup的设计初衷相违背了，此时WaitGroup就会直接panic。
下面通过说明使用的注意事项，来间接介绍WaitGroup的异常验证逻辑。
4.使用注意事项

4.1 Add方法和Done方法需要成对出现

下面是一个Add方法和Done方法没有成对出现的例子，此时Add方法调多了，此时计数器永远大于0，Wait 方法会一直阻塞等待。

1. package main
3. import (
4.     "fmt"
5.     "sync"
6. )
8. func main() {
9.     var wg sync.WaitGroup
11.     wg.Add(2)
13.     go func() {
14.         defer wg.Done()
15.         fmt.Println("Goroutine 1")
16.     }()
18.     go func() {
19.         fmt.Println("Goroutine 2")
20.     }()
22.     wg.Wait()
24.     fmt.Println("All goroutines finished")
25. }

复制代码

在上述代码中，我们调用了wg.Add(2)，但只调用了一次wg.Done()。这会导致counter的值大于0，因此调用wg.Wait()会被永久阻塞，不会继续向下继续执行。
还有另外一种情况时Done方法调用多了，此时任务计数器counter的值为负数，从WaitGroup设计的语意来看，就是需要等待完成的任务数为负数，这个不符合预期，此时将会直接panic

1. package main
3. import (
4.     "fmt"
5.     "sync"
6. )
8. func main() {
9.     var wg sync.WaitGroup
10.     wg.Add(1)
12.     go func() {
13.         fmt.Println("Goroutine 1 started")
14.         wg.Done() // 第一次调用Done方法
15.         wg.Done() // 第二次调用Done方法
16.         fmt.Println("Goroutine 1 completed")
17.     }()
19.     wg.Wait()
20.     fmt.Println("All goroutines completed")
21. }

复制代码

在上面的例子中，我们启动了一个goroutine，第一次调用Add方法，counter的值变为1，在第14行调用Done，此时计数器的值变为0，此时等待中的goroutine将会被唤醒。在第15行又调用了一次Done方法，当counter减小为0时，再次调用Done方法会导致panic，因为此时waitGroup的计数器已经为0，再次减少将导致负数计数，这是不被允许的。
所以在调用Done方法时，需要保证每次调用都与Add方法的调用一一对应，否则会导致程序出现错误。
4.2 在所有任务都已经添加之后，才调用Wait方法进行等待

WaitGroup的设计初衷就是为了等待一组操作完成后再执行下一步操作。所以，如果在所有任务添加之前，便调用Wait方法进行等待，此时有可能会导致等待协程提前被唤醒，执行下一步操作，而尚未添加的任务则不会被等待，这违反了WaitGroup的设计初衷，也不符合预期。下面是一个简单的例子:
[code]package mainimport (        "fmt"        "sync"        "time")func main() {        var wg sync.WaitGroup        for i := 1; i