Go语言学习之WaitGroup用法详解

第Go语言学习之WaitGroup用法详解目录前言小试牛刀总览底层实现结构体AddDoneWait易错点总结

前言

在前面的文章中，我们使用过WaitGroup进行任务编排，Go语言中的WaitGroup和Java中的CyclicBarrier、CountDownLatch非常类似。比如我们有一个主任务在执行，执行到某一点时需要并行执行三个子任务，并且需要等到三个子任务都执行完后，再继续执行主任务。那我们就需要设置一个检查点，使主任务一直阻塞在这，等三个子任务执行完后再放行。

说明：本文中的示例，均是基于Go1.1764位机器

小试牛刀

我们先来个简单的例子，看下WaitGroup是怎么使用的。示例中使用Add(5)表示我们有5个子任务，然后起了5个协程去完成任务，主协程使用Wait()方法等待子协程执行完毕，输出一共等待的时间。

funcmain(){

varwaitGroupsync.WaitGroup

start:=time.Now()

waitGroup.Add(5)

fori:=0;ii++{

gofunc(){

deferwaitGroup.Done()

time.Sleep(time.Second)

fmt.Println("done")

waitGroup.Wait()

fmt.Println(time.Now().Sub(start).Seconds())

1.000306089

*/

总览

WaitGroup一共有三个方法：

(wg*WaitGroup)Add(deltaint)

(wg*WaitGroup)Done()

(wg*WaitGroup)Wait()

Add方法用于设置WaitGroup的计数值，可以理解为子任务的数量Done方法用于将WaitGroup的计数值减一，可以理解为完成一个子任务Wait方法用于阻塞调用者，直到WaitGroup的计数值为0，即所有子任务都完成

正常来说，我们使用的时候，需要先确定子任务的数量，然后调用Add()方法传入相应的数量，在每个子任务的协程中，调用Done()，需要等待的协程调用Wait()方法，状态流转如下图：

底层实现

结构体

typeWaitGroupstruct{

noCopynoCopy//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

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

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

func(wg*WaitGroup)Add(deltaint){

//拿到计数值等待者变量statep和信号量semap

statep,semap:=wg.state()

//计数值加上delta:statep的前四个字节是计数值，因此将delta前移32位

state:=atomic.AddUint64(statep,uint64(delta)32)

//计数值

v:=int32(state32)

//等待者数量

w:=uint32(state)

//如果加上delta之后，计数值变为负数，不合法，panic

ifv0{

panic("sync:negativeWaitGroupcounter")

//delta0v==int32(delta):表示从0开始添加计数值

//w!=0：表示已经有了等待者

//说明在添加计数值的时候，同时添加了等待者，非法操作。添加等待者需要在添加计数值之后

ifw!=0delta0v==int32(delta){

panic("sync:WaitGroupmisuse:AddcalledconcurrentlywithWait")

//v0:计数值不等于0，不需要唤醒等待者，直接返回

//w==0:没有等待者，不需要唤醒，直接返回

ifv0||w==0{

return

//再次检查数据是否一致

if*statep!=state{

panic("sync:WaitGroupmisuse:AddcalledconcurrentlywithWait")

//到这里说明计数值为0，且等待者大于0，需要唤醒所有的等待者，并把系统置为初始状态（0状态）

//将计数值和等待者数量都置为0

*statep=0

//唤醒等待者

for;w!=0;w--{

runtime_Semrelease(semap,false,0)

}

Done

//完成一个任务，将计数值减一，当计数值减为0时，需要唤醒所有的等待者

func(wg*WaitGroup)Done(){

wg.Add(-1)

}

Wait

//调用Wait方法会被阻塞，直到计数值变为0

func(wg*WaitGroup)Wait(){

//获取计数、等待数和信号量

statep,semap:=wg.state()

for{

state:=atomic.LoadUint64(statep)

//计数值

v:=int32(state32)

//等待者数量

w:=uint32(state)

//计数值数量为0，直接返回，无需等待

ifv==0{

return

//到这里说明计数值数量大于0

//增加等待者数量：这里会有竞争，比如多个Wait调用，或者在同时调用Add方法，增加不成功会继续for循环

ifatomic.CompareAndSwapUint64(statep,state,state+1){

//增加成功后，阻塞在信号量这里，等待被唤醒

runtime_Semacquire(semap)

//被唤醒的时候，应该是0状态。如果重用WaitGroup，需要等Wait返回

if*statep!=0{

panic("sync:WaitGroupisreusedbeforepreviousWaithasreturned")

return

}

易错点

上面分析源码可以看到几个会产生panic的点，这也是我们使用WaitGroup需要注意的地方

1.计数值变为负数

调用Add时参数值传负数

funcmain(){

varwgsync.WaitGroup

wg.Add(1)

wg.Add(-1)

wg.Add(-1)

}

多次调用Done方法

funcmain(){

varwgsync.WaitGroup

wg.Add(1)

gofunc(){

fmt.Println("test")

wg.Done()

wg.Done()

time.Sleep(time.Second)

wg.Wait()

}

2.Add和Wait并发调用

Add和Wait并发调用，有可能达不到我们预期的效果，甚至panic。如下示例中，我们想要等待3个子任务都执行完后再执行主任务，但实际情况可能是子任务还没起来，主任务就继续往下执行了。

funcdoSomething(wg*sync.WaitGroup){

wg.Add(1)

fmt.Println("dosomething")

deferwg.Done()

funcmain(){

varwgsync.WaitGroup

fori:=0;ii++{

godoSomething(wg)

wg.Wait()

fmt.Println("main")

//main

//dosomething

//dosomething

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

funcdoSomething(wg*sync.WaitGroup){

deferwg.Done()

fmt.Println("dosomething")

funcmain(){

varwgsync.WaitGroup

wg.Add(3)

fori:=0;ii++{

godoSomething(wg)

wg.Wait()

fmt.Println("main")

//dosomething

//dosomething

//dosomething

//main

3.没有等Wait返回，就重用WaitGroup

funcmain(){

varwgsync.WaitGroup

wg.Add(1)

gofunc(){

fmt.Println("dosomething")

wg.Done()

wg.Add(1)

wg.Wait()

}

4.复制使用

我们知道Go语言中的参数传递，都是值传递，就会产生复制操作。因此在向函数传递WaitGroup时，使用指针进行操作。

//错误使用方式，没有使用指针

funcdoSomething(wgsync.WaitGroup){

fmt.Println("dosomething")

deferwg.Done()

funcmain(){

varwgsync.WaitGroup

wg.Add(3)

fori:=0;ii++{

//这里没使用指针，wg状态一直不会改变，导致W