说明

golang通过waitgroup来实现并发控制,用法跟java的CountDownLatch 效果一样

WaitGroup 的使用场景和方法

我们通过goroutine运行一个或者一组任务,需要关心这组任务执行完了进行通知

WaitGroup 如同它的字面意思，就是等待一组 goroutine 运行完成，主要有三个方法组成：

Add(delta int) ：添加任务数
Wait()：阻塞等待所有任务的完成
Done()：完成任务

package main

import (
    "fmt"
    "sync"
    "time"
)

func worker(wg *sync.WaitGroup) {
    doSomething()
    wg.Done() // 2.1、完成任务
}

func main() {
    var wg sync.WaitGroup
    wg.Add(5) // 1、添加 5 个任务
    for i := 1; i <= 5; i++ {
        go worker(&wg) // 2、每个任务并发执行
    }
    wg.Wait() // 3、阻塞等待所有任务完成

}

WaitGroup 源码分析

sync/waitgroup.go

type WaitGroup struct {
    noCopy noCopy
    state1 [3]uint32
}

noCopy
其中，noCopy 表示 WaitGroup 是不可复制的。那么什么叫不可复制呢？

举个例子，当我们对函数参数定义了这个不可复制的类型时，开发者只能通过指针来传递函数参数。而规定使用指针传递又有什么好处呢？

好处在于如果有多个函数都定义了这个不可复制的参数时，那么这多个函数参数就可以共用同一个指针变量，来同步执行结果。而 WaitGroup 就是需要这样的约束规定。

state1 字段
接下来我们来看看 WaitGroup 的 state1 字段。state1 是一个包含了 counter 总数、 waiter 等待数、sema 信号量的 uint32 数组。

每当有 goroutine 调用了 Wait() 方法阻塞等待时，就会对 waiter 数量 + 1，然后等待信号量的唤起通知。

当我们调用 Add() 方法时，就会对 state1 的 counter 数量 + 1。

当调用 Done() 方法时就会对 counter 数量 -1。

直到 counter == 0 时就可以通过信号量唤起对应 waiter 数量的 goroutine 了，也就是唤起刚刚阻塞等待的 goroutine 们

ADD方法

func (wg *WaitGroup) Add(delta int) {
    statep, semap := wg.state()
    if race.Enabled { // 此处是 go 的竞争检测，可以不用关心
        _ = *statep
        if delta < 0 {
            race.ReleaseMerge(unsafe.Pointer(wg))
        }
        race.Disable()
        defer race.Enable()
    }
    state := atomic.AddUint64(statep, uint64(delta)<<32)
    v := int32(state >> 32) // 获取 counter
    w := uint32(state) // 获取 waiter
    if race.Enabled && delta > 0 && v == int32(delta) { // go 的竞争检测，可以不用关心
        race.Read(unsafe.Pointer(semap))
    }
    if v < 0 {
        panic("sync: negative WaitGroup counter")
    }
    if w != 0 && delta > 0 && v == int32(delta) {
        panic("sync: WaitGroup misuse: Add called concurrently with Wait")
    }
    if v > 0 || w == 0 { // counter > 0：还有任务在执行；waiter == 0 表示没有在阻塞等待的 goroutine
        return
    }
    if *statep != state {
        panic("sync: WaitGroup misuse: Add called concurrently with Wait")
    }
    // 执行到此处相当于 countr = 0，即所有的任务都已执行完，需要唤起等待的 goroutine了
    *statep = 0
    for ; w != 0; w-- {
        runtime_Semrelease(semap, false, 0)
    }
}

Done

func (wg *WaitGroup) Done() {
    wg.Add(-1) // 直接调用 Add 方法 对 counter -1
}

Wait方法

func (wg *WaitGroup) Wait() {
    statep, semap := wg.state()
    if race.Enabled { // go 的竞争检测，可以不用关心
        _ = *statep
        race.Disable()
    }
    for {
        state := atomic.LoadUint64(statep)
        v := int32(state >> 32)
        w := uint32(state)
        if v == 0 {
            // counter 为 0， 不需要再等待了。
            if race.Enabled {
                race.Enable()
                race.Acquire(unsafe.Pointer(wg))
            }
            return
        }
        // waiters 数目 +1.
        if atomic.CompareAndSwapUint64(statep, state, state+1) {
            if race.Enabled && w == 0 {
                race.Write(unsafe.Pointer(semap)) // go 的竞争检测，可以不用关心
            }
            runtime_Semacquire(semap) // 阻塞等待唤起
            if *statep != 0 {
                panic("sync: WaitGroup is reused before previous Wait has returned")
            }
            if race.Enabled {
                race.Enable()
                race.Acquire(unsafe.Pointer(wg))
            }
            return
        }
    }
}