GosyncWaitGroup使用深入理解

第GosyncWaitGroup使用深入理解目錄基本介紹使用源碼分析AddDoneWait注意事項(xiàng)

基本介紹

WaitGroup是go用來(lái)做任務(wù)編排的一個(gè)并發(fā)原語(yǔ)，它要解決的就是并發(fā)-等待的問(wèn)題：

當(dāng)有一個(gè)goroutineA在檢查點(diǎn)（checkpoint）等待一組goroutine全部完成，如果這些goroutine還沒(méi)全部完成，goroutineA就會(huì)阻塞在檢查點(diǎn)，直到所有g(shù)oroutine都完成后才能繼續(xù)執(zhí)行

試想如果沒(méi)有WaitGroup，想要在協(xié)程A等到其他協(xié)程執(zhí)行完成后能立馬執(zhí)行，只能不斷輪詢(xún)其他協(xié)程是否執(zhí)行完畢，這樣的問(wèn)題是：

及時(shí)性差：輪詢(xún)間隔越高，及時(shí)性越差無(wú)謂的空輪訓(xùn)，浪費(fèi)系統(tǒng)資源

而用WaitGroup時(shí)，協(xié)程A只用阻塞，直到其他協(xié)程執(zhí)行完畢后，再通知協(xié)程A

其他語(yǔ)言也提供了類(lèi)似的工具，例如Java的CountDownLatch

使用

Waitgroup提供了3個(gè)方法：

func(wg*WaitGroup)Add(deltaint)

func(wg*WaitGroup)Done()

func(wg*WaitGroup)Wait()

Add：增加計(jì)數(shù)值

Done：減少計(jì)數(shù)值

Wait：調(diào)用這個(gè)方法的goroutine會(huì)一直阻塞，直到WaitGroup的計(jì)數(shù)值變?yōu)?

源碼分析

typeWaitGroupstruct{

//避免復(fù)制

noCopynoCopy

//64位環(huán)境下，高32位是計(jì)數(shù)值，低32位記錄waiter的數(shù)量

state1uint64

//用于信號(hào)量

state2uint32

Add

func(wg*WaitGroup)Add(deltaint){

//獲取狀態(tài)值，信號(hào)量

statep,semap:=wg.state()

//將參數(shù)delta左32位，加到statep中，即給計(jì)數(shù)值加上delta

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

//加后的計(jì)數(shù)值

v:=int32(state32)

//waiter的數(shù)量

w:=uint32(state)

//加后不能是負(fù)值

ifv0{

panic("sync:negativeWaitGroupcounter")

//有waiter的情況下，當(dāng)前協(xié)程又加了計(jì)數(shù)值，panic

//即有waiter的情況下，不能再給waitgroup增加計(jì)數(shù)值了

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

panic("sync:WaitGroupmisuse:AddcalledconcurrentlywithWait")

//如果加完后v大于0，或者加完后v等于0，但沒(méi)有等待者，直接返回

ifv0||w==0{

return

//接下來(lái)就是v等于0，且w大于0的情況

//再次檢查是否有Add和Wait并發(fā)調(diào)用的情況

if*statep!=state{

panic("sync:WaitGroupmisuse:AddcalledconcurrentlywithWait")

//將計(jì)數(shù)值和waiter數(shù)量清0

*statep=0

//喚醒所有的waiter

for;w!=0;w--{

runtime_Semrelease(semap,false,0)

因?yàn)閟tate高32位保存計(jì)數(shù)值，因此需要將參數(shù)delta左移32位后加到state上才正確

如果加完后v大于0，或者加完后v等于0，但沒(méi)有等待者，直接返回

v大于0：表示自己不是最后一個(gè)調(diào)用Done的協(xié)程，不用自己來(lái)釋放waiter，直接返回v等于0，但沒(méi)有等待者：因?yàn)闆](méi)有等待者，也就不用釋放等待者，也直接返回

否則就是v等于0，且w大于0的情況：

自己是最后一個(gè)調(diào)用Done的，且還有等待者，那就喚醒所有等待者

Done

Done內(nèi)部調(diào)用Add，只是參數(shù)傳-1，表示減少計(jì)數(shù)值

func(wg*WaitGroup)Done(){

wg.Add(-1)

Wait

func(wg*WaitGroup)Wait(){

statep,semap:=wg.state()

for{

state:=atomic.LoadUint64(statep)

//v：計(jì)數(shù)值

v:=int32(state32)

w:=uint32(state)

//如果計(jì)數(shù)值為0，自己不需要等到，直接返回

ifv==0{

return

//增加waiter計(jì)數(shù)值

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

//自己在信號(hào)量上阻塞

runtime_Semacquire(semap)

//檢查Waitgroup是否在wait返回前被重用

if*statep!=0{

panic("sync:WaitGroupisreusedbeforepreviousWaithasreturned")

return

如果計(jì)數(shù)值為0，當(dāng)前不需要阻塞，直接返回

否則將waiter數(shù)量加1，如果添加成功，就把自己阻塞到信號(hào)量上

被喚醒時(shí)，如果statep不為0，表示該waitgroup是否在wait返回前被重用了，panic

注意事項(xiàng)

通過(guò)源碼分析可以看出，Waitgroup有以下使用注意事項(xiàng)：

計(jì)數(shù)器的值必須大于等于0：

一開(kāi)始調(diào)用Add時(shí)，不能傳負(fù)數(shù)

調(diào)用Done的次數(shù)不能過(guò)多，導(dǎo)致超過(guò)了WaitGroup的計(jì)數(shù)值

因此使用WaitGroup的正確姿勢(shì)是，預(yù)先確定好WaitGroup的計(jì)數(shù)值，然后調(diào)用相同次數(shù)的Done完成相應(yīng)的任務(wù)

要保證在期望的Add調(diào)用完成后，再調(diào)用Wait，否則Wait發(fā)現(xiàn)計(jì)數(shù)值為0時(shí)不會(huì)阻塞

最好在一個(gè)協(xié)程中，按順序先調(diào)Add，再調(diào)Wait

需要重用時(shí)，需要在前一組調(diào)用Wait結(jié)束后，再開(kāi)始新一輪的使用

WaitGroup是可以重用