一文详解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 started\n", id)
              time.Sleep(time.Duration(id) * time.Second) 
              fmt.Printf("Goroutine %d finished\n", 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
 }