1. 前言
上一篇介紹了 Go并發(fā)控制--Channel
使用channel來控制子協(xié)程的優(yōu)點(diǎn)是實(shí)現(xiàn)簡單,缺點(diǎn)是當(dāng)需要大量創(chuàng)建協(xié)程時(shí)就需要有相同數(shù)量的channel,而且對于子協(xié)程繼續(xù)派生出來的協(xié)程不方便控制。
2. 使用WaitGroup控制
WaitGroup,可理解為Wait-Goroutine-Group,即等待一組goroutine結(jié)束。比如某個(gè)goroutine需要等待其他幾個(gè)goroutine全部完成,那么使用WaitGroup可以輕松實(shí)現(xiàn)。
2.1 使用場景
下面程序展示了一個(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 信號量
信號量是Unix系統(tǒng)提供的一種保護(hù)共享資源的機(jī)制,用于防止多個(gè)線程同時(shí)訪問某個(gè)資源
可簡單理解為信號量為一個(gè)數(shù)值:
- 當(dāng)信號量>0時(shí),表示資源可用,獲取信號量時(shí)系統(tǒng)自動(dòng)將信號量減1;
- 當(dāng)信號量==0時(shí),表示資源暫不可用,獲取信號量時(shí),當(dāng)前線程會(huì)進(jìn)入睡眠,當(dāng)信號量為正時(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è)信號量,而state實(shí)際上是兩個(gè)計(jì)數(shù)器:
- counter: 當(dāng)前還未執(zhí)行結(jié)束的goroutine計(jì)數(shù)器
- waiter count: 等待goroutine-group結(jié)束的goroutine數(shù)量,即有多少個(gè)等候者
- semaphore: 信號量
考慮到字節(jié)是否對齊,三者出現(xiàn)的位置不同,為簡單起見,依照字節(jié)已對齊情況下,三者在內(nèi)存中的位置如下所示:
WaitGroup對外提供三個(gè)接口:
- Add(delta int): 將delta值加到counter中
- Wait(): waiter遞增1,并阻塞等待信號量semaphore
- Done(): counter遞減1,按照waiter數(shù)值釋放相應(yīng)次數(shù)信號量
下面分別介紹這三個(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ù)值釋放等量的信號量,把等待的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為正,說明不需要釋放信號量,直接退出
//如果waiter為零,說明沒有等待者,也不需要釋放信號量,直接退出
if v > 0 || w == 0 {
return
}
//此時(shí),counter一定等于0,而waiter一定大于0(內(nèi)部維護(hù)waiter,不會(huì)出現(xiàn)小于0的情況),
//先把counter置為0,再釋放waiter個(gè)數(shù)的信號量
*statep = 0
for ; w != 0; w-- {
runtime_Semrelease(semap, false) //釋放信號量,執(zhí)行一次釋放一個(gè),喚醒一個(gè)等待者
}
}
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,累加可能會(huì)失敗,失敗后通過for loop下次重試
if atomic.CompareAndSwapUint64(statep, state, state+1) {
runtime_Semacquire(semap) //累加成功后,等待信號量喚醒自己
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é)束的場景,其內(nèi)部維護(hù)兩個(gè)計(jì)數(shù)器,這里把它們稱為“工作協(xié)程”計(jì)數(shù)器和“坐等協(xié)程”計(jì)數(shù)器,
WaitGroup對外提供的三個(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í)檢測到死鎖會(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**上場了
到此這篇關(guān)于Go并發(fā)控制--WaitGroup篇的文章就介紹到這了,更多相關(guān)Go并發(fā)控制WaitGroup內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家!
您可能感興趣的文章:- Golang 實(shí)現(xiàn)分片讀取http超大文件流和并發(fā)控制
- Go 并發(fā)控制context實(shí)現(xiàn)原理剖析(小結(jié))
- Django Channels 實(shí)現(xiàn)點(diǎn)對點(diǎn)實(shí)時(shí)聊天和消息推送功能
- 基于django channel實(shí)現(xiàn)websocket的聊天室的方法示例
- Go并發(fā)控制Channel使用場景分析