?

go版本 ：1.10.3

原理實(shí)現(xiàn)：信號(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í)被喚醒

?

WaitGroup的定義

type WaitGroup struct {noCopy noCopy // noCopy用來標(biāo)記不可復(fù)制,只能用指針傳遞,保證全局唯一.其實(shí)即使復(fù)制了,編譯,運(yùn)行都沒問題,只有用go vet檢測時(shí)才會(huì)顯示出錯(cuò)誤// 只需要64位,即8個(gè)字節(jié),其中高32位是counter值,低32位值是waiter值// 不直接使用uint64,是因?yàn)閡int64的原子操作需要64位系統(tǒng),而32位系統(tǒng)下,可能會(huì)出現(xiàn)崩潰// 所以這里用byte數(shù)組來實(shí)現(xiàn),32位系統(tǒng)下4字節(jié)對齊,64位系統(tǒng)下8字節(jié)對齊,所以申請12個(gè)字節(jié),其中必定有8個(gè)字節(jié)是符合8字節(jié)對齊的,下面的state()函數(shù)中有進(jìn)行判斷state1 [12]bytesema uint32 // 信號(hào)量 } // 得到counter值（uint64的高32位），waiter值（uint64的低32位） func (wg *WaitGroup) state() *uint64 {// 根據(jù)state1的起始地址分析,若是8字節(jié)對齊的,則直接用前8個(gè)字節(jié)作為*uint64類型// 若不是,說明是4字節(jié)對齊,則后移4個(gè)字節(jié)后,這樣必為8字節(jié)對齊,然后取后面8個(gè)字節(jié)作為*uint64類型if uintptr(unsafe.Pointer(&wg.state1))%8 == 0 {return (*uint64)(unsafe.Pointer(&wg.state1))} else {return (*uint64)(unsafe.Pointer(&wg.state1[4]))} }

增加或減少counter的值

func (wg *WaitGroup) Add(delta int) {//當(dāng)前的 counter值 waiter值statep := wg.state()// 把delta值加到counter上state := atomic.AddUint64(statep, uint64(delta)<<32)v := int32(state >> 32) // counter值w := uint32(state) // waiter值// counter為負(fù),則panicif v < 0 {panic("sync: negative WaitGroup counter")}// waiter值不為0,累加后的counter值和delta相等,說明Add()和Wait()同時(shí)調(diào)用了,panic,因?yàn)檎_的做法是先Add()后Wait()if w != 0 && delta > 0 && v == int32(delta) {panic("sync: WaitGroup misuse: Add called concurrently with Wait")}// 正常Add()后的情況// 1. counter > 0,說明還不需要釋放信號(hào)量,返回// 2. waiter = 0,說明沒有等待的goruntine,也不需要釋放信號(hào)量,返回if v > 0 || w == 0 {return}// 下面是counter==0,且w>0的情況// 現(xiàn)在若原state和新state不等,有以下兩種可能// 1. add和wait同時(shí)調(diào)用// 2. counter已經(jīng)為0,但waiter還為正值,這種情況永遠(yuǎn)不可能觸發(fā)信號(hào)量了// 都是出錯(cuò)了if *statep != state {panic("sync: WaitGroup misuse: Add called concurrently with Wait")}// 把counter,waiter都置為0，因?yàn)橐呀?jīng)觸發(fā)信號(hào)，通知所有等待的goroutine即可，此時(shí)不可以再Add()或者Wait()了*statep = 0// 原子地遞增信號(hào)量,并通知等待的goroutinefor ; w != 0; w-- {runtime_Semrelease(&wg.sema, false)} } // Done函數(shù)即簡單將counter值減1 func (wg *WaitGroup) Done() {wg.Add(-1) }

增加waiter值

func (wg *WaitGroup) Wait() {//當(dāng)前的 counter值 waiter值statep := wg.state()// 一直等待,直到無需等待或信號(hào)量觸發(fā)時(shí),才返回for {state := atomic.LoadUint64(statep)v := int32(state >> 32) // counter值w := uint32(state) // waiter值// 若counter值為0,說明所有g(shù)oroutine都退出了,無需等待,直接返回即可if v == 0 {return}// 原子地增加waiter的值,CAS方法,外面有for循環(huán)會(huì)一直嘗試,保證多個(gè)goroutine同時(shí)調(diào)用Wait()也能正確累加waiterif atomic.CompareAndSwapUint64(statep, state, state+1) {// 一直等待信號(hào)量sema,直到>0,信號(hào)量觸發(fā),然后以原子的方式遞減它runtime_Semacquire(&wg.sema)// 看上面的Add()函數(shù)，觸發(fā)信號(hào)量前會(huì)先將counter和waiter置0，所以此時(shí)必定為0// 若不為0，說明WaitGroup此時(shí)又被執(zhí)行Add()或者Wait()操作了，應(yīng)panicif *statep != 0 {panic("sync: WaitGroup is reused before previous Wait has returned")}// 可以返回了return}} }

提示

• Add()操作必須早于Wait(), 否則會(huì)panic
• Add()設(shè)置的值必須與實(shí)際等待的goroutine個(gè)數(shù)一致，否則會(huì)panic
• WaitGroup只可保持一份，不可拷貝給其他變量，否則會(huì)造成意想不到的BUG

總結(jié)

以上是生活随笔為你收集整理的go sync.WaitGroup源码分析的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。