Go进阶并发控制channel和WaitGroup的实现

package main

import (
    "fmt"
    "gomodule/data"
    _ "gomodule/pubsub"
    "time"
)

func main() {
    //创建一个有10个元素的channel
    channel := make([]chan int, 10)
    for i := 0; i < 10; i++ {
       //切片中放入一个channel
       channel[i] = make(chan int)
       //启动协程
       go Process(channel[i])
    }

    for i, ch := range channel {
       <-ch
       fmt.Println("Routine ", i, "quit!")
    }
}

func Process(ch chan int) {
    time.Sleep(1 * time.Second)
    //管道写入一个元素代表协程结束.
    ch <- 1
}

package main

import (
    "fmt"
    "gomodule/data"
    _ "gomodule/pubsub"
    "sync"
    "time"
)

func main() {
    var wg sync.WaitGroup
    //计数器.数值是goroutine的个数.
    wg.Add(2)
    go func() {
       time.Sleep(1 * time.Second)
       fmt.Println("goroutine 1 finish")
       //计数器减一.
       wg.Done()
    }()

    go func() {
       time.Sleep(2 * time.Second)
       fmt.Println("goroutine 2 finish")
       wg.Done()
    }()

    //主goroutine阻塞等待计数器变为0.
    wg.Wait()
    fmt.Println("all goroutine finish")
}

type WaitGroup struct {
    noCopy noCopy

    state atomic.Uint64 // high 32 bits are counter, low 32 bits are waiter count.
    sema  uint32
}

func (wg *WaitGroup) Add(delta int) {
    if race.Enabled {
       if delta < 0 {
          // Synchronize decrements with Wait.
          race.ReleaseMerge(unsafe.Pointer(wg))
       }
       race.Disable()
       defer race.Enable()
    }
    state := wg.state.Add(uint64(delta) << 32)
    v := int32(state >> 32)
    w := uint32(state)
    if race.Enabled && delta > 0 && v == int32(delta) {
       // The first increment must be synchronized with Wait.
       // Need to model this as a read, because there can be
       // several concurrent wg.counter transitions from 0.
       race.Read(unsafe.Pointer(&wg.sema))
    }
    if v < 0 {
       panic("sync: negative WaitGroup counter")
    }
    if w != 0 && delta > 0 && v == int32(delta) {
       panic("sync: WaitGroup misuse: Add called concurrently with Wait")
    }
    if v > 0 || w == 0 {
       return
    }
    // This goroutine has set counter to 0 when waiters > 0.
    // Now there can't be concurrent mutations of state:
    // - Adds must not happen concurrently with Wait,
    // - Wait does not increment waiters if it sees counter == 0.
    // Still do a cheap sanity check to detect WaitGroup misuse.
    if wg.state.Load() != state {
       panic("sync: WaitGroup misuse: Add called concurrently with Wait")
    }
    // Reset waiters count to 0.
    wg.state.Store(0)
    for ; w != 0; w-- {
       runtime_Semrelease(&wg.sema, false, 0)
    }
}

func (wg *WaitGroup) Wait() {
    if race.Enabled {
       race.Disable()
    }
    for {
       state := wg.state.Load()
       v := int32(state >> 32)
       w := uint32(state)
       if v == 0 {
          // Counter is 0, no need to wait.
          if race.Enabled {
             race.Enable()
             race.Acquire(unsafe.Pointer(wg))
          }
          return
       }
       // Increment waiters count.
       if wg.state.CompareAndSwap(state, state+1) {
          if race.Enabled && w == 0 {
             // Wait must be synchronized with the first Add.
             // Need to model this is as a write to race with the read in Add.
             // As a consequence, can do the write only for the first waiter,
             // otherwise concurrent Waits will race with each other.
             race.Write(unsafe.Pointer(&wg.sema))
          }
          runtime_SemacquireWaitGroup(&wg.sema)
          if wg.state.Load() != 0 {
             panic("sync: WaitGroup is reused before previous Wait has returned")
          }
          if race.Enabled {
             race.Enable()
             race.Acquire(unsafe.Pointer(wg))
          }
          return
       }
    }
}

func (wg *WaitGroup) Done() {
    wg.Add(-1)
}