Go語言學習之WaitGroup用法詳解

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

前言

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

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

小試牛刀

我們先來個簡單的例子，看下WaitGroup是怎么使用的。示例中使用Add(5)表示我們有5個子任務，然后起了5個協(xié)程去完成任務，主協(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方法用于設置WaitGroup的計數值，可以理解為子任務的數量Done方法用于將WaitGroup的計數值減一，可以理解為完成一個子任務Wait方法用于阻塞調用者，直到WaitGroup的計數值為0，即所有子任務都完成

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

底層實現(xiàn)

結構體

typeWaitGroupstruct{

noCopynoCopy//noCopy字段標識，由于WaitGroup不能復制，方便工具檢測

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

}

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

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

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

//首地址是8的倍數時，前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方法用于添加一個計數值（負數相當于減），當計數值變?yōu)?后，Wait方法阻塞的所有等待者都會被釋放計數值變?yōu)樨摂凳欠欠ú僮?，產生panic當計數值為0時（初始狀態(tài)），Add方法不能和Wait方法并發(fā)調用，需要保證Add方法在Wait方法之前調用，否則會panic

func(wg*WaitGroup)Add(deltaint){

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

statep,semap:=wg.state()

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

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

//計數值

v:=int32(state32)

//等待者數量

w:=uint32(state)

//如果加上delta之后，計數值變?yōu)樨摂?，不合法，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，需要喚醒所有的等待者，并把系統(tǒng)置為初始狀態(tài)（0狀態(tài)）

//將計數值和等待者數量都置為0

*statep=0

//喚醒等待者

for;w!=0;w--{

runtime_Semrelease(semap,false,0)

}

Done

//完成一個任務，將計數值減一，當計數值減為0時，需要喚醒所有的等待者

func(wg*WaitGroup)Done(){

wg.Add(-1)

}

Wait

//調用Wait方法會被阻塞，直到計數值變?yōu)?

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方法，增加不成功會繼續(xù)for循環(huán)

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

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

runtime_Semacquire(semap)

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

if*statep!=0{

panic("sync:WaitGroupisreusedbeforepreviousWaithasreturned")

return

}

易錯點

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

1.計數值變?yōu)樨摂?/p>

調用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并發(fā)調用

Add和Wait并發(fā)調用，有可能達不到我們預期的效果，甚至panic。如下示例中，我們想要等待3個子任務都執(zhí)行完后再執(zhí)行主任務，但實際情況可能是子任務還沒起來，主任務就繼續(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

正確的使用方式，應該是在調用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狀態(tài)一直不會改變，導致W