1. 前言

上一篇介紹了 Go并發(fā)控制--Channel

使用channel來控制子協(xié)程的優(yōu)點(diǎn)是實(shí)現(xiàn)簡單，缺點(diǎn)是當(dāng)需要大量創(chuàng)建協(xié)程時(shí)就需要有相同數(shù)量的channel，而且對(duì)于子協(xié)程繼續(xù)派生出來的協(xié)程不方便控制。

2. 使用WaitGroup控制

WaitGroup，可理解為Wait-Goroutine-Group，即等待一組goroutine結(jié)束。比如某個(gè)goroutine需要等待其他幾個(gè)goroutine全部完成，那么使用WaitGroup可以輕松實(shí)現(xiàn)。

2.1 使用場(chǎng)景

下面程序展示了一個(gè)goroutine等待另外兩個(gè)goroutine結(jié)束的例子：

package main

import (
    "fmt"
    "time"
    "sync"
)

func main() {
    var wg sync.WaitGroup

    wg.Add(2) //設(shè)置計(jì)數(shù)器，數(shù)值即為goroutine的個(gè)數(shù)
    go func() {
        //Do some work
        time.Sleep(1*time.Second)

        fmt.Println("Goroutine 1 finished!")
        wg.Done() //goroutine執(zhí)行結(jié)束后將計(jì)數(shù)器減1
    }()

    go func() {
        //Do some work
        time.Sleep(2*time.Second)

        fmt.Println("Goroutine 2 finished!")
        wg.Done() //goroutine執(zhí)行結(jié)束后將計(jì)數(shù)器減1
    }()

    wg.Wait() //主goroutine阻塞等待計(jì)數(shù)器變?yōu)?
    fmt.Printf("All Goroutine finished!")
}

簡單的說，上面程序中wg內(nèi)部維護(hù)了一個(gè)計(jì)數(shù)器：

• 啟動(dòng)goroutine前將計(jì)數(shù)器通過Add(2)將計(jì)數(shù)器設(shè)置為待啟動(dòng)的goroutine個(gè)數(shù)。
• 啟動(dòng)goroutine后，使用Wait()方法阻塞自己，等待計(jì)數(shù)器變?yōu)?。
• 每個(gè)goroutine執(zhí)行結(jié)束通過Done()方法將計(jì)數(shù)器減1。
• 計(jì)數(shù)器變?yōu)?后，阻塞的goroutine被喚醒

其實(shí)WaitGroup也可以實(shí)現(xiàn)一組goroutine等待另一組goroutine，這有點(diǎn)像玩雜技，很容出錯(cuò)，如果不了解其實(shí)現(xiàn)原理更是如此。實(shí)際上，WaitGroup的實(shí)現(xiàn)源碼非常簡單。

2.2 信號(hào)量

信號(hào)量是Unix系統(tǒng)提供的一種保護(hù)共享資源的機(jī)制，用于防止多個(gè)線程同時(shí)訪問某個(gè)資源

可簡單理解為信號(hào)量為一個(gè)數(shù)值：

• 當(dāng)信號(hào)量>0時(shí)，表示資源可用，獲取信號(hào)量時(shí)系統(tǒng)自動(dòng)將信號(hào)量減1；
• 當(dāng)信號(hào)量==0時(shí)，表示資源暫不可用，獲取信號(hào)量時(shí)，當(dāng)前線程會(huì)進(jìn)入睡眠，當(dāng)信號(hào)量為正時(shí)被喚醒；
  

1.3 WaitGroup 數(shù)據(jù)結(jié)構(gòu)

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

type WaitGroup struct {
    state1 [3]uint32
}

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

• counter： 當(dāng)前還未執(zhí)行結(jié)束的goroutine計(jì)數(shù)器
• waiter count: 等待goroutine-group結(jié)束的goroutine數(shù)量，即有多少個(gè)等候者
• semaphore: 信號(hào)量

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

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

• Add(delta int): 將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.1 Add () 方法

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(delta int) {
    statep, semap := wg.state() //獲取state和semaphore地址指針

    state := atomic.AddUint64(statep, uint64(delta)32) //把delta左移32位累加到state，即累加到counter中
    v := int32(state >> 32) //獲取counter值
    w := uint32(state)      //獲取waiter值

    if v  0 {              //經(jīng)過累加后counter值變?yōu)樨?fù)值，panic
        panic("sync: negative WaitGroup counter")
    }

    //經(jīng)過累加后，此時(shí)，counter >= 0
    //如果counter為正，說明不需要釋放信號(hào)量，直接退出
    //如果waiter為零，說明沒有等待者，也不需要釋放信號(hào)量，直接退出
    if v > 0 || 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.2 Wait()

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

func (wg *WaitGroup) Wait() {
    statep, semap := wg.state() //獲取state和semaphore地址指針
    for {
        state := atomic.LoadUint64(statep) //獲取state值
        v := int32(state >> 32)            //獲取counter值
        w := uint32(state)                 //獲取waiter值
        if v == 0 {                        //如果counter值為0，說明所有g(shù)oroutine都退出了，不需要待待，直接返回
            return
        }

        // 使用CAS（比較交換算法）累加waiter，累加可能會(huì)失敗，失敗后通過for loop下次重試
        if atomic.CompareAndSwapUint64(statep, state, state+1) {
            runtime_Semacquire(semap) //累加成功后，等待信號(hào)量喚醒自己
            return
        }
    }
}

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

2.3.3 Done()

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(delta int)方法用于增加“工作協(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)panic

3. 總結(jié)

WaitGroup控制子協(xié)程的方式很簡單，且目的很明確，等待一組子協(xié)程執(zhí)行完畢再執(zhí)行主線程，但是當(dāng)子協(xié)程里面有子協(xié)程，子協(xié)程里面有其他的子協(xié)程時(shí)，這種并不知道有多少個(gè)子協(xié)程的情況下使用WaitGroup就很難，所以就需要****Context**上場(chǎng)了

到此這篇關(guān)于Go并發(fā)控制--WaitGroup篇的文章就介紹到這了,更多相關(guān)Go并發(fā)控制WaitGroup內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！