一文詳解go同步協程的必備工具WaitGroup

func main(){
    //首先主協程建立WaitGroup範例
    var wg sync.WaitGroup
    // 開始時呼叫Add方法表示有個任務開始執行
    wg.Add(1)
    go func() {
        // 開始執行...
        //完成之後,呼叫Done方法
        wg.Done()
    }()
    // 呼叫Wait()方法，等待所有協程完成任務
    wg.Wait()
    // 執行後續邏輯
}

package main
import (
    "fmt"
    "sync"
)
func main() {
    var wg sync.WaitGroup
    for i := 0; i < 5; i++ {
       wg.Add(1)
       go func(i int) {
          defer wg.Done()
          fmt.Printf("任務%d開始執行n", i)
          // 模擬協程任務執行一段時間
          time.Sleep(time.Duration(rand.Int() % 100))
          // 執行緒任務執行完成
          fmt.Printf("任務%d執行完畢n", i)
       }(i)
    }
    fmt.Println("主協程開始等待所有任務執行完成...")
    wg.Wait()
    fmt.Println("所有協程已經執行完畢...")
}

type WaitGroup struct {
   noCopy noCopy
   state1 uint64
   state2 uint32
}

func (wg *WaitGroup) Add(delta int) {
   // delta 的值可以為負數,Done方法便是通過Add(-1)來實現的
   // statep: 為state1的地址  semap: 為state2的地址
   statep, semap := wg.state()
   // 高32位元的值 加上 delta,增加任務計數器的值
   state := atomic.AddUint64(statep, uint64(delta)<<32)
   // v: 取高32位元資料,獲取到待完成任務數
   v := int32(state >> 32)
   // 取低32位元資料,獲取到等待執行緒的值
   w := uint32(state)
   // v > 0: 說明還有待完成的任務數,此時不應該喚醒等待協程
   // w = 0: 說明沒有協程在等待,此時可以直接退出
   if v > 0 || w == 0 {
      return
   }     
   // 此時v = 0,所有任務都完成了,喚醒等待協程
   *statep = 0
   for ; w != 0; w-- {
      runtime_Semrelease(semap, false, 0)
   }
}

// Done decrements the WaitGroup counter by one.
func (wg *WaitGroup) Done() {
   wg.Add(-1)
}

func (wg *WaitGroup) Wait() {
   // statep: 為state1的地址  semap: 為state2的地址
   statep, semap := wg.state()
   for {
      // 載入state1的值
      state := atomic.LoadUint64(statep)
      // v: 取高32位元資料,獲取到待完成任務數
      v := int32(state >> 32)
      // 沒有任務待執行,全部都完成了
      if v == 0 {
         return
      }
      // 增加waiter計數器的值
      if atomic.CompareAndSwapUint64(statep, state, state+1) {
          // 等待被喚醒
         runtime_Semacquire(semap)
         return
      }
   }
}

package main
import (
    "fmt"
    "sync"
)
func main() {
    var wg sync.WaitGroup
    wg.Add(2)
    go func() {
        defer wg.Done()
        fmt.Println("Goroutine 1")
    }()
    go func() {
        fmt.Println("Goroutine 2")
    }()
    wg.Wait()
    fmt.Println("All goroutines finished")
}

package main
import (
    "fmt"
    "sync"
)
func main() {
    var wg sync.WaitGroup
    wg.Add(1)
    go func() {
        fmt.Println("Goroutine 1 started")
        wg.Done() // 第一次呼叫Done方法
        wg.Done() // 第二次呼叫Done方法
        fmt.Println("Goroutine 1 completed")
    }()
    wg.Wait()
    fmt.Println("All goroutines completed")
}

package main
import (
        "fmt"
        "sync"
        "time"
)
func main() {
        var wg sync.WaitGroup
        for i := 1; i <= 3; i++ {
           go func(id int) {
              wg.Add(1)
              defer wg.Done()
              fmt.Printf("Goroutine %d startedn", id)
              time.Sleep(time.Duration(id) * time.Second) 
              fmt.Printf("Goroutine %d finishedn", id)
           }(i)
        }
        // 不等待所有任務新增，就開始等待
        wg.Wait()
        fmt.Println("All goroutines finished")
        time.Sleep(10 * time.Second)
}

func (c *ClusterClient) ForEachMaster(
   ctx context.Context,
   fn func(ctx context.Context, client *Client) error,
) error {
   // 重新載入叢集狀態,以確保狀態資訊是最新的
   state, err := c.state.ReloadOrGet(ctx)
   if err != nil {
      return err
   }
   var wg sync.WaitGroup
   // 用於協程間通訊
   errCh := make(chan error, 1)
    // 獲取到redis叢集中所有的master節點
   for _, master := range state.Masters {
      // 啟動一個協程來執行該任務
      wg.Add(1)
      go func(node *clusterNode) {
         // 任務完成時,呼叫Done告知WaitGroup任務已完成
         defer wg.Done()
         err := fn(ctx, node.Client)
         if err != nil {
            select {
            case errCh <- err:
            default:
            }
         }
      }(master)
   }
   // 主協程等待所有任務完成
   wg.Wait()
   return nil
 }