快捷導(dǎo)航

Go sync WaitGroup使用深入理解

更新時(shí)間：2022年10月25日 14:23:06 作者：亞洲第一中鋒_哈

這篇文章主要為大家介紹了Go sync WaitGroup使用深入理解，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進(jìn)步，早日升職加薪

基本介紹

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

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

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

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

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

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

使用

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

func (wg *WaitGroup) Add(delta int)
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)?0

源碼分析

type WaitGroup struct {
   // 避免復(fù)制
   noCopy noCopy
   // 64位環(huán)境下，高32位是計(jì)數(shù)值，低32位記錄waiter的數(shù)量
   state1 uint64
   // 用于信號(hào)量
   state2 uint32
}

Add

func (wg *WaitGroup) Add(delta int) {
   // 獲取狀態(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(state >> 32)
   // waiter的數(shù)量
   w := uint32(state)
   // 加后不能是負(fù)值
   if v < 0 {
      panic( "sync: negative WaitGroup counter" )
   }
   // 有waiter的情況下，當(dāng)前協(xié)程又加了計(jì)數(shù)值，panic
   // 即有waiter的情況下，不能再給waitgroup增加計(jì)數(shù)值了
   if w != 0 && delta > 0 && v == int32(delta) {
      panic( "sync: WaitGroup misuse: Add called concurrently with Wait" )
   }
   // 如果加完后v大于0，或者加完后v等于0，但沒有等待者，直接返回
   if v > 0 || w == 0 {
      return
   }
   // 接下來就是v等于0，且w大于0的情況
   // 再次檢查是否有Add和Wait并發(fā)調(diào)用的情況
   if *statep != state {
      panic( "sync: WaitGroup misuse: Add called concurrently with Wait" )
   }
   // 將計(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，但沒有等待者，直接返回

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

否則就是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(state >> 32)
      w := uint32(state)
      // 如果計(jì)數(shù)值為0，自己不需要等到，直接返回
      if v == 0 {
         return
   }
      // 增加waiter計(jì)數(shù)值
 if atomic.CompareAndSwapUint64(statep, state, state+1) {
         // 自己在信號(hào)量上阻塞
         runtime_Semacquire(semap)
         // 檢查Waitgroup是否在wait返回前被重用
         if *statep != 0 {
            panic( "sync: WaitGroup is reused before previous Wait has returned" )
         }
         return
      }
   }
}

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

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

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