Go并發(fā)控制WaitGroup的使用場(chǎng)景分析

第Go并發(fā)控制WaitGroup的使用場(chǎng)景分析1.3WaitGroup數(shù)據(jù)結(jié)構(gòu)

源碼包中src/sync/waitgroup.go:WaitGroup定義了其數(shù)據(jù)結(jié)構(gòu)：

typeWaitGroupstruct{

state1[3]uint32

}

state1是個(gè)長度為3的數(shù)組，其中包含了state和一個(gè)信號(hào)量，而state實(shí)際上是兩個(gè)計(jì)數(shù)器：

counter：當(dāng)前還未執(zhí)行結(jié)束的goroutine計(jì)數(shù)器

waitercount:等待goroutine-group結(jié)束的goroutine數(shù)量，即有多少個(gè)等候者

semaphore:信號(hào)量

考慮到字節(jié)是否對(duì)齊，三者出現(xiàn)的位置不同，為簡單起見，依照字節(jié)已對(duì)齊情況下，三者在內(nèi)存中的位置如下所示：

WaitGroup對(duì)外提供三個(gè)接口：

Add(deltaint):將delta值加到counter中

Wait()：waiter遞增1，并阻塞等待信號(hào)量semaphore

Done()：counter遞減1，按照waiter數(shù)值釋放相應(yīng)次數(shù)信號(hào)量

下面分別介紹這三個(gè)函數(shù)的實(shí)現(xiàn)細(xì)節(jié)。

2.3.1Add()方法

Add()做了兩件事，一是把delta值累加到counter中，因?yàn)閐elta可以為負(fù)值，也就是說counter有可能變成0或負(fù)值，所以第二件事就是當(dāng)counter值變?yōu)?時(shí)，根據(jù)waiter數(shù)值釋放等量的信號(hào)量，把等待的goroutine全部喚醒，如果counter變?yōu)樨?fù)值，則panic.

Add()偽代碼如下：

func(wg*WaitGroup)Add(deltaint){

statep,semap:=wg.state()//獲取state和semaphore地址指針

state:=atomic.AddUint64(statep,uint64(delta)32)//把delta左移32位累加到state，即累加到counter中

v:=int32(state32)//獲取counter值

w:=uint32(state)//獲取waiter值

ifv0{//經(jīng)過累加后counter值變?yōu)樨?fù)值，panic

panic("sync:negativeWaitGroupcounter")

//經(jīng)過累加后，此時(shí)，counter=0

//如果counter為正，說明不需要釋放信號(hào)量，直接退出

//如果waiter為零，說明沒有等待者，也不需要釋放信號(hào)量，直接退出

ifv0||w==0{

return

//此時(shí)，counter一定等于0，而waiter一定大于0（內(nèi)部維護(hù)waiter，不會(huì)出現(xiàn)小于0的情況），

//先把counter置為0，再釋放waiter個(gè)數(shù)的信號(hào)量

*statep=0

for;w!=0;w--{

runtime_Semrelease(semap,false)//釋放信號(hào)量，執(zhí)行一次釋放一個(gè)，喚醒一個(gè)等待者

}

2.3.2Wait()

Wait()方法也做了兩件事，一是累加waiter,二是阻塞等待信號(hào)量

func(wg*WaitGroup)Wait(){

statep,semap:=wg.state()//獲取state和semaphore地址指針

for{

state:=atomic.LoadUint64(statep)//獲取state值

v:=int32(state32)//獲取counter值

w:=uint32(state)//獲取waiter值

ifv==0{//如果counter值為0，說明所有g(shù)oroutine都退出了，不需要待待，直接返回

return

//使用CAS（比較交換算法）累加waiter，累加可能會(huì)失敗，失敗后通過forloop下次重試

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

runtime_Semacquire(semap)//累加成功后，等待信號(hào)量喚醒自己

return

}

這里用到了CAS算法保證有多個(gè)goroutine同時(shí)執(zhí)行Wait()時(shí)也能正確累加waiter。

2.3.3Done()

Done()只做一件事，即把counter減1，我們知道Add()可以接受負(fù)值，所以Done實(shí)際上只是調(diào)用了Add(-1)。

源碼如下：

func(wg*WaitGroup)Done(){

wg.Add(-1)

}

Done()的執(zhí)行邏輯就轉(zhuǎn)到了Add()，實(shí)際上也正是最后一個(gè)完成的goroutine把等待者喚醒的。

2.4總結(jié)

簡單說來，WaitGroup通常用于等待一組“工作協(xié)程”結(jié)束的場(chǎng)景，其內(nèi)部維護(hù)兩個(gè)計(jì)數(shù)器，這里把它們稱為“工作協(xié)程”計(jì)數(shù)器和“坐等協(xié)程”計(jì)數(shù)器，

WaitGroup對(duì)外提供的三個(gè)方法分工非常明確：

Add(deltaint)方法用于增加“工作協(xié)程”計(jì)數(shù)，通常在啟動(dòng)新的“工作協(xié)程”之前調(diào)用；

Done()方法用于減少“工作協(xié)程”計(jì)數(shù)，每次調(diào)用遞減1，通常在“工作協(xié)程”內(nèi)部且在臨近返回之前調(diào)用；

Wait()方法用于增加“坐等協(xié)程”計(jì)數(shù)，通常在所有”工作協(xié)

Done()方法除了負(fù)責(zé)遞減“工作協(xié)程”計(jì)數(shù)以外，還會(huì)在“工作協(xié)程”計(jì)數(shù)變?yōu)?時(shí)檢查“坐等協(xié)程”計(jì)數(shù)器并把“坐等協(xié)程”喚醒。

需要注意

Done()方法遞減“工作協(xié)程”計(jì)數(shù)后，如果“工作協(xié)程”計(jì)數(shù)變成負(fù)數(shù)時(shí)，將會(huì)觸發(fā)panic，這就要求Add()方法調(diào)用要早于Done()方法。

也就是說代碼中，如果調(diào)用Done的次數(shù)多于Add的次數(shù)會(huì)產(chǎn)生painc

當(dāng)“工作協(xié)程”計(jì)數(shù)多于實(shí)際需要等待的“工作協(xié)程”數(shù)量時(shí)，“坐等協(xié)程”可能會(huì)永遠(yuǎn)無法被喚醒而產(chǎn)生列鎖，此時(shí)，Go運(yùn)行時(shí)檢測(cè)到死鎖會(huì)觸發(fā)panic

Add的添加的工作協(xié)程的數(shù)量，多于Done調(diào)用的次數(shù)，則會(huì)出現(xiàn)panic

當(dāng)“工作協(xié)程”計(jì)數(shù)小于實(shí)際需要等待的“工作協(xié)程”數(shù)量時(shí)，Done()會(huì)在“工作協(xié)程”計(jì)數(shù)變?yōu)樨?fù)數(shù)時(shí)觸發(fā)panic。

Add（）添加的工作協(xié)程個(gè)數(shù)小于Done調(diào)用的次數(shù)，會(huì)出現(xiàn)pani