关于go:Go语言-WaitGroup-详解

你必须十分致力，能力看起来毫不费力！

微信搜寻公众号[漫漫 Coding 路]，一起 From Zero To Hero !

在后面的文章中，咱们应用过 WaitGroup 进行工作编排，Go 语言中的 WaitGroup 和 Java 中的 CyclicBarrier、CountDownLatch 十分相似。比方咱们有一个主工作在执行，执行到某一点时须要并行执行三个子工作，并且须要等到三个子工作都执行完后，再继续执行主工作。那咱们就须要设置一个检查点，使主工作始终阻塞在这，等三个子工作执行完后再放行。

阐明：本文中的示例，均是基于 Go1.17 64 位机器

咱们先来个简略的例子，看下 WaitGroup 是怎么应用的。示例中应用 Add(5) 示意咱们有 5 个子工作，而后起了 5 个协程去实现工作，主协程应用 Wait() 办法期待子协程执行结束，输入一共期待的工夫。

func main() {
    var waitGroup sync.WaitGroup

    start := time.Now()
    waitGroup.Add(5)
    for i := 0; i < 5; i++ {go func() {defer waitGroup.Done()
            time.Sleep(time.Second)
            fmt.Println("done")
        }()}

    waitGroup.Wait()
    fmt.Println(time.Now().Sub(start).Seconds())
}

/*
done
done
done
done
done
1.000306089
*/

WaitGroup 一共有三个办法：

(wg *WaitGroup) Add(delta int)
(wg *WaitGroup) Done()
(wg *WaitGroup) Wait()

Add 办法用于设置 WaitGroup 的计数值，能够了解为子工作的数量
Done 办法用于将 WaitGroup 的计数值减一，能够了解为实现一个子工作
Wait 办法用于阻塞调用者，直到 WaitGroup 的计数值为 0，即所有子工作都实现

失常来说，咱们应用的时候，须要先确定子工作的数量，而后调用 Add() 办法传入相应的数量，在每个子工作的协程中，调用 Done()，须要期待的协程调用 Wait() 办法，状态流转如下图：

type WaitGroup struct {
    noCopy noCopy // noCopy 字段标识，因为 WaitGroup 不能复制，不便工具检测

    state1 [3]uint32  // 12 个字节，8 个字节标识 计数值和期待数量，4 个字节用于标识信号量
}

state1 是个复合字段，会拆分为两局部：64 位（8 个字节）的 statep 作为一个整体用于原子操作, 其中后面 4 个字节示意计数值，前面四个字节示意期待数量；残余 32 位（4 个字节）semap 用于标识信号量。

Go 语言中对于 64 位的变量进行原子操作，须要保障该变量是 64 位对齐 的，也就是要保障这 8 个字节的首地址是 8 的整数倍。因而当 state1 的首地址是 8 的整数倍时，取 前 8 个 字节作为 statep，后 4 个 字节作为 semap；当 state1 的首地址不是 8 的整数倍时，取 后 8 个 字节作为 statep，前 4 个 字节作为 semap。

func (wg *WaitGroup) state() (statep *uint64, semap *uint32) {
    
    // 首地址是 8 的倍数时，前 8 个字节为 statep, 后四个字节为 semap
    if uintptr(unsafe.Pointer(&wg.state1))%8 == 0 {return (*uint64)(unsafe.Pointer(&wg.state1)), &wg.state1[2]
    } else { 
        
    // 后 8 个字节为 statep, 前四个字节为 semap    
        return (*uint64)(unsafe.Pointer(&wg.state1[1])), &wg.state1[0]
    }
}

Add 办法用于增加一个计数值（正数相当于减），当计数值变为 0 后，Wait 办法阻塞的所有期待者都会被开释
计数值变为正数是非法操作，产生 panic
当计数值为 0 时（初始状态），Add 办法不能和 Wait 办法并发调用，须要保障 Add 办法在 Wait 办法 之前 调用，否则会 panic

func (wg *WaitGroup) Add(delta int) {

    // 拿到计数值期待者变量 statep 和 信号量 semap
    statep, semap := wg.state()

    // 计数值加上 delta: statep 的前四个字节是计数值，因而将 delta 前移 32 位
    state := atomic.AddUint64(statep, uint64(delta)<<32)

    // 计数值
    v := int32(state >> 32)

    // 期待者数量
    w := uint32(state)

    // 如果加上 delta 之后，计数值变为正数，不非法，panic
    if v < 0 {panic("sync: negative WaitGroup counter")
    }

    // delta > 0 && v == int32(delta) : 示意从 0 开始增加计数值
    // w!=0：示意曾经有了期待者
    // 阐明在增加计数值的时候，同时增加了期待者，非法操作。增加期待者须要在增加计数值之后
    if w != 0 && delta > 0 && v == int32(delta) {panic("sync: WaitGroup misuse: Add called concurrently with Wait")
    }

    // v>0 : 计数值不等于 0，不须要唤醒期待者，间接返回
    // w==0: 没有期待者，不须要唤醒，间接返回
    if v > 0 || w == 0 {return}

    // 再次检查数据是否统一
    if *statep != state {panic("sync: WaitGroup misuse: Add called concurrently with Wait")
    }

    // 到这里阐明计数值为 0，且期待者大于 0，须要唤醒所有的期待者，并把零碎置为初始状态（0 状态）// 将计数值和期待者数量都置为 0
    *statep = 0

    // 唤醒期待者
    for ; w != 0; w-- {runtime_Semrelease(semap, false, 0)
    }
}

// 实现一个工作，将计数值减一，当计数值减为 0 时，须要唤醒所有的期待者
func (wg *WaitGroup) Done() {wg.Add(-1)
}

// 调用 Wait 办法会被阻塞，直到 计数值 变为 0
func (wg *WaitGroup) Wait() {

    // 获取计数、期待数和信号量
    statep, semap := wg.state()

    for {state := atomic.LoadUint64(statep)

        // 计数值
        v := int32(state >> 32)

        // 期待者数量
        w := uint32(state)

        // 计数值数量为 0，间接返回，无需期待
        if v == 0 {return}

        // 到这里阐明计数值数量大于 0
        // 减少期待者数量：这里会有竞争，比方多个 Wait 调用，或者在同时调用 Add 办法，减少不胜利会持续 for 循环
        if atomic.CompareAndSwapUint64(statep, state, state+1) {
            // 减少胜利后，阻塞在信号量这里，期待被唤醒
            runtime_Semacquire(semap)

            // 被唤醒的时候，应该是 0 状态。如果重用 WaitGroup，须要等 Wait 返回
            if *statep != 0 {panic("sync: WaitGroup is reused before previous Wait has returned")
            }
            return
        }
    }
}

下面剖析源码能够看到几个会产生 panic 的点，这也是咱们应用 WaitGroup 须要留神的中央

计数值变为正数

调用 Add 时参数值传正数

func main() {
var wg sync.WaitGroup
wg.Add(1)
wg.Add(-1)
wg.Add(-1)
}

屡次调用 Done 办法

func main() {
    var wg sync.WaitGroup

    wg.Add(1)
    go func() {fmt.Println("test")
        wg.Done()
        wg.Done()}()

    time.Sleep(time.Second)
    wg.Wait()}

Add 和 Wait 并发调用

Add 和 Wait 并发调用，有可能达不到咱们预期的成果，甚至 panic。如下示例中，咱们想要期待 3 个子工作都执行完后再执行主工作，但理论状况可能是子工作还没起来，主工作就持续往下执行了。

func doSomething(wg *sync.WaitGroup) {wg.Add(1)
    fmt.Println("do  something")
    defer wg.Done()}

func main() {
    var wg sync.WaitGroup
    for i := 0; i < 3; i++ {go doSomething(&wg)
    }

    wg.Wait()
    fmt.Println("main")
}

//main
//do  something
//do  something

正确的应用形式，应该是在调用 Wait 前先调用 Add

func doSomething(wg *sync.WaitGroup) {defer wg.Done()
    fmt.Println("do  something")
}

func main() {
    var wg sync.WaitGroup
    wg.Add(3)
    for i := 0; i < 3; i++ {go doSomething(&wg)
    }

    wg.Wait()
    fmt.Println("main")
}

//do  something
//do  something
//do  something
//main

没有等 Wait 返回，就重用 WaitGroup

func main() {
    var wg sync.WaitGroup

    wg.Add(1)
    go func() {fmt.Println("do something")
        wg.Done()
        wg.Add(1)
    }()

    wg.Wait()}

复制应用

咱们晓得 Go 语言中的参数传递，都是值传递，就会产生复制操作。因而在向函数传递 WaitGroup 时，应用指针 进行操作。

// 谬误应用形式，没有应用指针
func doSomething(wg sync.WaitGroup) {fmt.Println("do  something")
    defer wg.Done()}

func main() {
    var wg sync.WaitGroup
    wg.Add(3)
    for i := 0; i < 3; i++ {
        // 这里没应用指针，wg 状态始终不会扭转，导致 Wait 始终阻塞
        go doSomething(wg)
    }

    wg.Wait()
    fmt.Println("main")
}

本篇文章，咱们通过源码 + 示例的形式，一起学习了 sync.WaitGroup 实现逻辑，同时也给出了一些留神点，只有做到如下操作，就不会呈现问题：

保障 Add 在 Wait 前调用
Add 中不传递正数
工作实现后不要遗记调用 Done 办法，倡议应用 defer wg.Done()
不要复制应用 WaitGroup，函数传递时应用指针传递
尽量不复用 WaigGroup，缩小出问题的危险

集体博客: https://lifelmy.github.io/

微信公众号：漫漫 Coding 路

关于go:Go语言-WaitGroup-详解

前言

小试牛刀

总览

底层实现

构造体

Add

Done

Wait

易错点

总结

更多

Just My Socks（注册教程内含优惠码）

关于go:Go语言-WaitGroup-详解

前言

小试牛刀

总览

底层实现

构造体

Add

Done

Wait

易错点

总结

更多

Just My Socks（注册教程 内含优惠码）

Just My Socks（注册教程内含优惠码）