Go語言學(xué)習(xí)之WaitGroup用法詳解

第Go語言學(xué)習(xí)之WaitGroup用法詳解目錄前言小試牛刀總覽底層實現(xiàn)結(jié)構(gòu)體AddDoneWait易錯點總結(jié)

前言

在前面的文章中，我們使用過WaitGroup進(jìn)行任務(wù)編排，Go語言中的WaitGroup和Java中的CyclicBarrier、CountDownLatch非常類似。比如我們有一個主任務(wù)在執(zhí)行，執(zhí)行到某一點時需要并行執(zhí)行三個子任務(wù)，并且需要等到三個子任務(wù)都執(zhí)行完后，再繼續(xù)執(zhí)行主任務(wù)。那我們就需要設(shè)置一個檢查點，使主任務(wù)一直阻塞在這，等三個子任務(wù)執(zhí)行完后再放行。

說明：本文中的示例，均是基于Go1.1764位機(jī)器

小試牛刀

我們先來個簡單的例子，看下WaitGroup是怎么使用的。示例中使用Add(5)表示我們有5個子任務(wù)，然后起了5個協(xié)程去完成任務(wù)，主協(xié)程使用Wait()方法等待子協(xié)程執(zhí)行完畢，輸出一共等待的時間。

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方法用于設(shè)置WaitGroup的計數(shù)值，可以理解為子任務(wù)的數(shù)量Done方法用于將WaitGroup的計數(shù)值減一，可以理解為完成一個子任務(wù)Wait方法用于阻塞調(diào)用者，直到WaitGroup的計數(shù)值為0，即所有子任務(wù)都完成

正常來說，我們使用的時候，需要先確定子任務(wù)的數(shù)量，然后調(diào)用Add()方法傳入相應(yīng)的數(shù)量，在每個子任務(wù)的協(xié)程中，調(diào)用Done()，需要等待的協(xié)程調(diào)用Wait()方法，狀態(tài)流轉(zhuǎn)如下圖：

底層實現(xiàn)

結(jié)構(gòu)體

typeWaitGroupstruct{

noCopynoCopy//noCopy字段標(biāo)識，由于WaitGroup不能復(fù)制，方便工具檢測

state1[3]uint32//12個字節(jié)，8個字節(jié)標(biāo)識計數(shù)值和等待數(shù)量，4個字節(jié)用于標(biāo)識信號量

}

state1是個復(fù)合字段，會拆分為兩部分：64位（8個字節(jié)）的statep作為一個整體用于原子操作,其中前面4個字節(jié)表示計數(shù)值，后面四個字節(jié)表示等待數(shù)量；剩余32位（4個字節(jié)）semap用于標(biāo)識信號量。

Go語言中對于64位的變量進(jìn)行原子操作，需要保證該變量是64位對齊的，也就是要保證這8個字節(jié)的首地址是8的整數(shù)倍。因此當(dāng)state1的首地址是8的整數(shù)倍時，取前8個字節(jié)作為statep，后4個字節(jié)作為semap；當(dāng)state1的首地址不是8的整數(shù)倍時，取后8個字節(jié)作為statep，前4個字節(jié)作為semap。

func(wg*WaitGroup)state()(statep*uint64,semap*uint32){

//首地址是8的倍數(shù)時，前8個字節(jié)為statep,后四個字節(jié)為semap

ifuintptr(unsafe.Pointer(wg.state1))%8==0{

return(*uint64)(unsafe.Pointer(wg.state1)),wg.state1[2]

}else{

//后8個字節(jié)為statep,前四個字節(jié)為semap

return(*uint64)(unsafe.Pointer(wg.state1[1])),wg.state1[0]

}

Add

Add方法用于添加一個計數(shù)值（負(fù)數(shù)相當(dāng)于減），當(dāng)計數(shù)值變?yōu)?后，Wait方法阻塞的所有等待者都會被釋放計數(shù)值變?yōu)樨?fù)數(shù)是非法操作，產(chǎn)生panic當(dāng)計數(shù)值為0時（初始狀態(tài)），Add方法不能和Wait方法并發(fā)調(diào)用，需要保證Add方法在Wait方法之前調(diào)用，否則會panic

func(wg*WaitGroup)Add(deltaint){

//拿到計數(shù)值等待者變量statep和信號量semap

statep,semap:=wg.state()

//計數(shù)值加上delta:statep的前四個字節(jié)是計數(shù)值，因此將delta前移32位

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

//計數(shù)值

v:=int32(state32)

//等待者數(shù)量

w:=uint32(state)

//如果加上delta之后，計數(shù)值變?yōu)樨?fù)數(shù)，不合法，panic

ifv0{

panic("sync:negativeWaitGroupcounter")

//delta0v==int32(delta):表示從0開始添加計數(shù)值

//w!=0：表示已經(jīng)有了等待者

//說明在添加計數(shù)值的時候，同時添加了等待者，非法操作。添加等待者需要在添加計數(shù)值之后

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

panic("sync:WaitGroupmisuse:AddcalledconcurrentlywithWait")

//v0:計數(shù)值不等于0，不需要喚醒等待者，直接返回

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

ifv0||w==0{

return

//再次檢查數(shù)據(jù)是否一致

if*statep!=state{

panic("sync:WaitGroupmisuse:AddcalledconcurrentlywithWait")

//到這里說明計數(shù)值為0，且等待者大于0，需要喚醒所有的等待者，并把系統(tǒng)置為初始狀態(tài)（0狀態(tài)）

//將計數(shù)值和等待者數(shù)量都置為0

*statep=0

//喚醒等待者

for;w!=0;w--{

runtime_Semrelease(semap,false,0)

}

Done

//完成一個任務(wù)，將計數(shù)值減一，當(dāng)計數(shù)值減為0時，需要喚醒所有的等待者

func(wg*WaitGroup)Done(){

wg.Add(-1)

}

Wait

//調(diào)用Wait方法會被阻塞，直到計數(shù)值變?yōu)?

func(wg*WaitGroup)Wait(){

//獲取計數(shù)、等待數(shù)和信號量

statep,semap:=wg.state()

for{

state:=atomic.LoadUint64(statep)

//計數(shù)值

v:=int32(state32)

//等待者數(shù)量

w:=uint32(state)

//計數(shù)值數(shù)量為0，直接返回，無需等待

ifv==0{

return

//到這里說明計數(shù)值數(shù)量大于0

//增加等待者數(shù)量：這里會有競爭，比如多個Wait調(diào)用，或者在同時調(diào)用Add方法，增加不成功會繼續(xù)for循環(huán)

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

//增加成功后，阻塞在信號量這里，等待被喚醒

runtime_Semacquire(semap)

//被喚醒的時候，應(yīng)該是0狀態(tài)。如果重用WaitGroup，需要等Wait返回

if*statep!=0{

panic("sync:WaitGroupisreusedbeforepreviousWaithasreturned")

return

}

易錯點

上面分析源碼可以看到幾個會產(chǎn)生panic的點，這也是我們使用WaitGroup需要注意的地方

1.計數(shù)值變?yōu)樨?fù)數(shù)

調(diào)用Add時參數(shù)值傳負(fù)數(shù)

funcmain(){

varwgsync.WaitGroup

wg.Add(1)

wg.Add(-1)

wg.Add(-1)

}

多次調(diào)用Done方法

funcmain(){

varwgsync.WaitGroup

wg.Add(1)

gofunc(){

fmt.Println("test")

wg.Done()

wg.Done()

time.Sleep(time.Second)

wg.Wait()

}

2.Add和Wait并發(fā)調(diào)用

Add和Wait并發(fā)調(diào)用，有可能達(dá)不到我們預(yù)期的效果，甚至panic。如下示例中，我們想要等待3個子任務(wù)都執(zhí)行完后再執(zhí)行主任務(wù)，但實際情況可能是子任務(wù)還沒起來，主任務(wù)就繼續(xù)往下執(zhí)行了。

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

正確的使用方式，應(yīng)該是在調(diào)用Wait前先調(diào)用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.復(fù)制使用

我們知道Go語言中的參數(shù)傳遞，都是值傳遞，就會產(chǎn)生復(fù)制操作。因此在向函數(shù)傳遞WaitGroup時，使用指針進(jìn)行操作。

//錯誤使用方式，沒有使用指針

funcdoSomething(wgsync.WaitGroup){

fmt.Println("dosomething")

deferwg.Done()

funcmain(){

varwgsync.WaitGroup

wg.Add(3)

fori:=0;ii++{

//這里沒使用指針，wg狀態(tài)一直不會改變，導(dǎo)致W