1. 前言
上一篇介紹了 Go并發(fā)控制--Channel
使用channel來控制子協(xié)程的優(yōu)點是實現(xiàn)簡單,缺點是當(dāng)需要大量創(chuàng)建協(xié)程時就需要有相同數(shù)量的channel,而且對于子協(xié)程繼續(xù)派生出來的協(xié)程不方便控制。
2. 使用WaitGroup控制
WaitGroup,可理解為Wait-Goroutine-Group,即等待一組goroutine結(jié)束。比如某個goroutine需要等待其他幾個goroutine全部完成,那么使用WaitGroup可以輕松實現(xiàn)。
2.1 使用場景
下面程序展示了一個goroutine等待另外兩個goroutine結(jié)束的例子:
package main
import (
"fmt"
"time"
"sync"
)
func main() {
var wg sync.WaitGroup
wg.Add(2) //設(shè)置計數(shù)器,數(shù)值即為goroutine的個數(shù)
go func() {
//Do some work
time.Sleep(1*time.Second)
fmt.Println("Goroutine 1 finished!")
wg.Done() //goroutine執(zhí)行結(jié)束后將計數(shù)器減1
}()
go func() {
//Do some work
time.Sleep(2*time.Second)
fmt.Println("Goroutine 2 finished!")
wg.Done() //goroutine執(zhí)行結(jié)束后將計數(shù)器減1
}()
wg.Wait() //主goroutine阻塞等待計數(shù)器變?yōu)?
fmt.Printf("All Goroutine finished!")
}
簡單的說,上面程序中wg內(nèi)部維護(hù)了一個計數(shù)器:
- 啟動goroutine前將計數(shù)器通過Add(2)將計數(shù)器設(shè)置為待啟動的goroutine個數(shù)。
- 啟動goroutine后,使用Wait()方法阻塞自己,等待計數(shù)器變?yōu)?。
- 每個goroutine執(zhí)行結(jié)束通過Done()方法將計數(shù)器減1。
- 計數(shù)器變?yōu)?后,阻塞的goroutine被喚醒
其實WaitGroup也可以實現(xiàn)一組goroutine等待另一組goroutine,這有點像玩雜技,很容出錯,如果不了解其實現(xiàn)原理更是如此。實際上,WaitGroup的實現(xiàn)源碼非常簡單。
2.2 信號量
信號量是Unix系統(tǒng)提供的一種保護(hù)共享資源的機(jī)制,用于防止多個線程同時訪問某個資源
可簡單理解為信號量為一個數(shù)值:
- 當(dāng)信號量>0時,表示資源可用,獲取信號量時系統(tǒng)自動將信號量減1;
- 當(dāng)信號量==0時,表示資源暫不可用,獲取信號量時,當(dāng)前線程會進(jìn)入睡眠,當(dāng)信號量為正時被喚醒;
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是個長度為3的數(shù)組,其中包含了state和一個信號量,而state實際上是兩個計數(shù)器:
- counter: 當(dāng)前還未執(zhí)行結(jié)束的goroutine計數(shù)器
- waiter count: 等待goroutine-group結(jié)束的goroutine數(shù)量,即有多少個等候者
- semaphore: 信號量
考慮到字節(jié)是否對齊,三者出現(xiàn)的位置不同,為簡單起見,依照字節(jié)已對齊情況下,三者在內(nèi)存中的位置如下所示:
WaitGroup對外提供三個接口:
- Add(delta int): 將delta值加到counter中
- Wait(): waiter遞增1,并阻塞等待信號量semaphore
- Done(): counter遞減1,按照waiter數(shù)值釋放相應(yīng)次數(shù)信號量
下面分別介紹這三個函數(shù)的實現(xiàn)細(xì)節(jié)。
2.3.1 Add () 方法
Add()做了兩件事,一是把delta值累加到counter中,因為delta可以為負(fù)值,也就是說counter有可能變成0或負(fù)值,所以第二件事就是當(dāng)counter值變?yōu)?時,根據(jù)waiter數(shù)值釋放等量的信號量,把等待的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)過累加后,此時,counter >= 0
//如果counter為正,說明不需要釋放信號量,直接退出
//如果waiter為零,說明沒有等待者,也不需要釋放信號量,直接退出
if v > 0 || w == 0 {
return
}
//此時,counter一定等于0,而waiter一定大于0(內(nèi)部維護(hù)waiter,不會出現(xiàn)小于0的情況),
//先把counter置為0,再釋放waiter個數(shù)的信號量
*statep = 0
for ; w != 0; w-- {
runtime_Semrelease(semap, false) //釋放信號量,執(zhí)行一次釋放一個,喚醒一個等待者
}
}
2.3.2 Wait()
Wait()方法也做了兩件事,一是累加waiter, 二是阻塞等待信號量
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,累加可能會失敗,失敗后通過for loop下次重試
if atomic.CompareAndSwapUint64(statep, state, state+1) {
runtime_Semacquire(semap) //累加成功后,等待信號量喚醒自己
return
}
}
}
這里用到了CAS算法保證有多個goroutine同時執(zhí)行Wait()時也能正確累加waiter。
2.3.3 Done()
Done()只做一件事,即把counter減1,我們知道Add()可以接受負(fù)值,所以Done實際上只是調(diào)用了Add(-1)。
源碼如下:
func (wg *WaitGroup) Done() {
wg.Add(-1)
}
Done()的執(zhí)行邏輯就轉(zhuǎn)到了Add(),實際上也正是最后一個完成的goroutine把等待者喚醒的。
2.4 總結(jié)
簡單說來,WaitGroup通常用于等待一組“工作協(xié)程”結(jié)束的場景,其內(nèi)部維護(hù)兩個計數(shù)器,這里把它們稱為“工作協(xié)程”計數(shù)器和“坐等協(xié)程”計數(shù)器,
WaitGroup對外提供的三個方法分工非常明確:
- Add(delta int)方法用于增加“工作協(xié)程”計數(shù),通常在啟動新的“工作協(xié)程”之前調(diào)用;
- Done()方法用于減少“工作協(xié)程”計數(shù),每次調(diào)用遞減1,通常在“工作協(xié)程”內(nèi)部且在臨近返回之前調(diào)用;
- Wait()方法用于增加“坐等協(xié)程”計數(shù),通常在所有”工作協(xié)
Done()方法除了負(fù)責(zé)遞減“工作協(xié)程”計數(shù)以外,還會在“工作協(xié)程”計數(shù)變?yōu)?時檢查“坐等協(xié)程”計數(shù)器并把“坐等協(xié)程”喚醒。
需要注意
- Done()方法遞減“工作協(xié)程”計數(shù)后,如果“工作協(xié)程”計數(shù)變成負(fù)數(shù)時,將會觸發(fā)panic,這就要求Add()方法調(diào)用要早于Done()方法。
- 也就是說代碼中,如果調(diào)用Done的次數(shù)多于Add的次數(shù)會產(chǎn)生painc
- 當(dāng)“工作協(xié)程”計數(shù)多于實際需要等待的“工作協(xié)程”數(shù)量時,“坐等協(xié)程”可能會永遠(yuǎn)無法被喚醒而產(chǎn)生列鎖,此時,Go運行時檢測到死鎖會觸發(fā)panic
- Add的添加的工作協(xié)程的數(shù)量,多于Done調(diào)用的次數(shù),則會出現(xiàn)panic
- 當(dāng)“工作協(xié)程”計數(shù)小于實際需要等待的“工作協(xié)程”數(shù)量時,Done()會在“工作協(xié)程”計數(shù)變?yōu)樨?fù)數(shù)時觸發(fā)panic。
- Add()添加的工作協(xié)程個數(shù)小于Done調(diào)用的次數(shù),會出現(xiàn)panic
3. 總結(jié)
WaitGroup控制子協(xié)程的方式很簡單,且目的很明確,等待一組子協(xié)程執(zhí)行完畢再執(zhí)行主線程,但是當(dāng)子協(xié)程里面有子協(xié)程,子協(xié)程里面有其他的子協(xié)程時,這種并不知道有多少個子協(xié)程的情況下使用WaitGroup就很難,所以就需要****Context**上場了
到此這篇關(guān)于Go并發(fā)控制--WaitGroup篇的文章就介紹到這了,更多相關(guān)Go并發(fā)控制WaitGroup內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家!
您可能感興趣的文章:- Golang 實現(xiàn)分片讀取http超大文件流和并發(fā)控制
- Go 并發(fā)控制context實現(xiàn)原理剖析(小結(jié))
- Django Channels 實現(xiàn)點對點實時聊天和消息推送功能
- 基于django channel實現(xiàn)websocket的聊天室的方法示例
- Go并發(fā)控制Channel使用場景分析