关于面试:深度解析sync-WaitGroup源码及其实现原理

1次阅读

共计 3564 个字符，预计需要花费 9 分钟才能阅读完成。

WaitGroup 介绍
WaitGroup 的实现
- Add
- Done
- Wait

waitGroup，也是在 go 语言并发中比拟罕用的语法，所以在这里咱们一起分析 waitGroup 的应用形式及其源码解读。

WaitGroup 也是 sync 包下一份子，用来解决工作编排的一个并发原语。它次要解决了并发 - 期待问题：比方当初有三个 goroutine，别离为goroutineA，goroutineB，goroutineC，而goroutineA 须要期待 goroutineB 和goroutineC这一组 goroutine 全副执行结束后，才能够执行后续业务逻辑。此时就能够应用 WaitGroup 轻松解决。

在这个场景中，goroutineA为主 goroutine，goroutineB和 goroutineC 为子 goroutine。goroutineA则须要在 检查点 (checkout point) 期待goroutineB 和goroutineC全副执行结束，如果在执行工作的 goroutine 还没全副实现，那么 goroutineA 就会阻塞在检查点，直到所有 goroutine 都实现后能力继续执行。

代码实现：

package main

import (
  "fmt"
  "sync"
)

func goroutineB(wg *sync.WaitGroup) {defer wg.Done()
  fmt.Println("goroutineB Execute")
  time.Sleep(time.Second)
}

func goroutineC(wg *sync.WaitGroup) {defer wg.Done()
  fmt.Println("goroutineC Execute")
  time.Sleep(time.Second)
}

func main() {
  var wg sync.WaitGroup
  wg.Add(2)
  go goroutineB(&wg)
  go goroutineC(&wg)
  wg.Wait()
  fmt.Println("goroutineB and goroutineC finished...")
}

运行后果:

goroutineC Execute
goroutineB Execute
goroutineB and goroutineC finished...

上述就是WaitGroup 的简略操作，它的语法也是比较简单，提供了三个办法，如下所示：

func (wg *WaitGroup) Add(delta int)
func (wg *WaitGroup) Done()
func (wg *WaitGroup) Wait()

Add：用来设置 WaitGroup 的计数值(子 goroutine 的数量)
Done：用来将 WaitGroup 的计数值减 1，起始就是调用 Add(-1)
Wait：调用这个办法的 goroutine 会始终阻塞，直到 WaitGroup 的技术值变为 0

接下来，咱们进行分析 WaitGroup 的源码实现，让其无处可遁，它源码比拟少，除去正文，也就几十行，对老手来说也是一种不错的抉择。

首先，咱们看看 WaitGroup 的数据结构，它包含了一个 noCopy 的辅助字段，一个具备复合意义的 state1 字段。

noCopy 的辅助字段：次要就是辅助 vet 工具查看是否通过 copy 赋值这个 WaitGroup 实例。我会在前面和你详细分析这个字段
state1：具备复合意义的字段，蕴含 WaitGroup 计数值，阻塞在检查点的主 gooutine 和信号量

type WaitGroup struct {
    // 防止复制应用的一个技巧，能够通知 vet 工具违反了复制应用的规定
    noCopy noCopy
    // 64bit(8bytes)的值分成两段，高 32bit 是计数值，低 32bit 是 waiter 的计数
    // 另外 32bit 是用作信号量的
    // 因为 64bit 值的原子操作须要 64bit 对齐，然而 32bit 编译器不反对，所以数组中的元素在不同的架构中不一样，具体解决看上面的办法
    // 总之，会找到对齐的那 64bit 作为 state，其余的 32bit 做信号量
    state1 [3]uint32
}


// 失去 state 的地址和信号量的地址
func (wg *WaitGroup) state() (statep *uint64, semap *uint32) {if uintptr(unsafe.Pointer(&wg.state1))%8 == 0 {
        // 如果地址是 64bit 对齐的，数组前两个元素做 state，后一个元素做信号量
        return (*uint64)(unsafe.Pointer(&wg.state1)), &wg.state1[2]
    } else {
        // 如果地址是 32bit 对齐的，数组后两个元素用来做 state，它能够用来做 64bit 的原子操作，第一个元素 32bit 用来做信号量
        return (*uint64)(unsafe.Pointer(&wg.state1[1])), &wg.state1[0]
    }
}

因为对 64 位整数的原子操作要求整数的地址是 64 位对齐的，所以针对 64 位和 32 位环境的 state 字段的组成是不一样的。

在 64 位环境下，state1 的第一个元素是 waiter 数，第二个元素是 WaitGroup 的计数值，第三个元素是信号量。

在 32 位环境下，如果 state1 不是 64 位对齐的地址，那么 state1 的第一个元素是信号量，后两个元素别离是 waiter 数和计数值。

接下里，咱们一一看 Add 办法、Done 办法、Wait 办法的实现原理。

Add 办法实现思路：

Add 办法次要操作的 state1 字段中计数值局部。当 Add 办法被调用时，首先会将 delta 参数值左移 32 位 (计数值在高 32 位)，而后外部通过原子操作将这个值加到计数值上。须要留神的是，delta 的取值范畴可正可负，因为调用 Done() 办法时，外部通过 Add(-1)办法实现的。

代码实现如下：

func (wg *WaitGroup) Add(delta int) {
  // statep 示意 wait 数和计数值
  // 低 32 位示意 wait 数，高 32 位示意计数值
   statep, semap := wg.state()
   // uint64(delta)<<32 将 delta 左移 32 位
    // 因为高 32 位示意计数值，所以将 delta 左移 32，减少到技术上
   state := atomic.AddUint64(statep, uint64(delta)<<32)
   // 以后计数值
   v := int32(state >> 32)
   // 阻塞在检查点的 wait 数
   w := uint32(state)
   if v > 0 || w == 0 {return}
   
   // 如果计数值 v 为 0 并且 waiter 的数量 w 不为 0，那么 state 的值就是 waiter 的数量
    // 将 waiter 的数量设置为 0，因为计数值 v 也是 0, 所以它们俩的组合 *statep 间接设置为 0 即可。此时须要并唤醒所有的 waiter
   *statep = 0
   for ; w != 0; w-- {runtime_Semrelease(semap, false, 0)
   }
}

外部就是调用 Add(-1)办法，这里就不细讲了。

// Done 办法理论就是计数器减 1
func (wg *WaitGroup) Done() {wg.Add(-1)
}

wait 实现思路：

一直查看 state 值。如果其中的计数值为零，则阐明所有的子 goroutine 已全副执行结束，调用者不用期待，间接返回。如果计数值大于零，阐明此时还有工作没有实现，那么调用者变成期待者，须要退出 wait 队列，并且阻塞本人。

代码实现如下：

func (wg *WaitGroup) Wait() {
   // statep 示意 wait 数和计数值
   // 低 32 位示意 wait 数，高 32 位示意计数值
   statep, semap := wg.state()
   for {state := atomic.LoadUint64(statep)
      // 将 state 右移 32 位，示意以后计数值
      v := int32(state >> 32)
      // w 示意 waiter 期待值
      w := uint32(state)
      if v == 0 {
         // 如果以后计数值为零，示意以后子 goroutine 已全副执行结束，则间接返回
         return
      }
      // 否则应用原子操作将 state 值加一。if atomic.CompareAndSwapUint64(statep, state, state+1) {
         // 阻塞休眠期待
         runtime_Semacquire(semap)
         // 被唤醒，不再阻塞，返回
         return
      }
   }
}

到此，waitGroup 的根本应用和实现原理已介绍结束了，置信大家已有不一样的播种，咱们下期见。