你必须十分致力,能力看起来毫不费力!
微信搜寻公众号[ 漫漫Coding路 ],一起From Zero To Hero !
前言
在后面的文章中,咱们应用过 WaitGroup
进行工作编排,Go语言中的 WaitGroup
和 Java 中的 CyclicBarrier
、CountDownLatch
十分相似。比方咱们有一个主工作在执行,执行到某一点时须要并行执行三个子工作,并且须要等到三个子工作都执行完后,再继续执行主工作。那咱们就须要设置一个检查点,使主工作始终阻塞在这,等三个子工作执行完后再放行。
阐明:本文中的示例,均是基于Go1.17 64位机器
小试牛刀
咱们先来个简略的例子,看下 WaitGroup
是怎么应用的。示例中应用 Add(5)
示意咱们有 5个 子工作,而后起了 5个 协程去实现工作,主协程应用 Wait()
办法期待 子协程执行结束,输入一共期待的工夫。
func main() { var waitGroup sync.WaitGroup start := time.Now() waitGroup.Add(5) for i := 0; i < 5; i++ { go func() { defer waitGroup.Done() time.Sleep(time.Second) fmt.Println("done") }() } waitGroup.Wait() fmt.Println(time.Now().Sub(start).Seconds())}/*donedonedonedonedone1.000306089*/
总览
WaitGroup 一共有三个办法:
(wg *WaitGroup) Add(delta int)(wg *WaitGroup) Done()(wg *WaitGroup) Wait()
Add
办法用于设置 WaitGroup 的计数值,能够了解为子工作的数量Done
办法用于将 WaitGroup 的计数值减一,能够了解为实现一个子工作Wait
办法用于阻塞调用者,直到 WaitGroup 的计数值为0,即所有子工作都实现
失常来说,咱们应用的时候,须要先确定子工作的数量,而后调用 Add() 办法传入相应的数量,在每个子工作的协程中,调用 Done(),须要期待的协程调用 Wait() 办法,状态流转如下图:
底层实现
构造体
type WaitGroup struct { noCopy noCopy // noCopy 字段标识,因为 WaitGroup 不能复制,不便工具检测 state1 [3]uint32 // 12个字节,8个字节标识 计数值和期待数量,4个字节用于标识信号量}
state1
是个复合字段,会拆分为两局部: 64位
(8个字节)的 statep
作为一个整体用于原子操作, 其中后面4个字节示意计数值,前面四个字节示意期待数量;残余 32位
(4个字节)semap
用于标识信号量。
Go语言中对于64位的变量进行原子操作,须要保障该变量是 64位对齐
的,也就是要保障这 8个字节 的首地址是 8 的整数倍。因而当 state1
的首地址是 8 的整数倍时,取前8个
字节作为 statep
,后4个
字节作为 semap
;当 state1
的首地址不是 8 的整数倍时,取后8个
字节作为 statep
,前4个
字节作为 semap
。
func (wg *WaitGroup) state() (statep *uint64, semap *uint32) { // 首地址是8的倍数时,前8个字节为 statep, 后四个字节为 semap if uintptr(unsafe.Pointer(&wg.state1))%8 == 0 { return (*uint64)(unsafe.Pointer(&wg.state1)), &wg.state1[2] } else { // 后8个字节为 statep, 前四个字节为 semap return (*uint64)(unsafe.Pointer(&wg.state1[1])), &wg.state1[0] }}
Add
Add
办法用于增加一个计数值(正数相当于减),当计数值变为0后,Wait
办法阻塞的所有期待者都会被开释- 计数值变为正数是非法操作,产生
panic
- 当计数值为0时(初始状态),
Add
办法不能和Wait
办法并发调用,须要保障Add
办法在Wait
办法之前
调用,否则会panic
func (wg *WaitGroup) Add(delta int) { // 拿到计数值期待者变量 statep 和 信号量 semap statep, semap := wg.state() // 计数值加上 delta: statep 的前四个字节是计数值,因而将 delta 前移 32位 state := atomic.AddUint64(statep, uint64(delta)<<32) // 计数值 v := int32(state >> 32) // 期待者数量 w := uint32(state) // 如果加上 delta 之后,计数值变为正数,不非法,panic if v < 0 { panic("sync: negative WaitGroup counter") } // delta > 0 && v == int32(delta) : 示意从 0 开始增加计数值 // w!=0 :示意曾经有了期待者 // 阐明在增加计数值的时候,同时增加了期待者,非法操作。增加期待者须要在增加计数值之后 if w != 0 && delta > 0 && v == int32(delta) { panic("sync: WaitGroup misuse: Add called concurrently with Wait") } // v>0 : 计数值不等于0,不须要唤醒期待者,间接返回 // w==0: 没有期待者,不须要唤醒,间接返回 if v > 0 || w == 0 { return } // 再次检查数据是否统一 if *statep != state { panic("sync: WaitGroup misuse: Add called concurrently with Wait") } // 到这里阐明计数值为0,且期待者大于0,须要唤醒所有的期待者,并把零碎置为初始状态(0状态) // 将计数值和期待者数量都置为0 *statep = 0 // 唤醒期待者 for ; w != 0; w-- { runtime_Semrelease(semap, false, 0) }}
Done
// 实现一个工作,将计数值减一,当计数值减为0时,须要唤醒所有的期待者func (wg *WaitGroup) Done() { wg.Add(-1)}
Wait
// 调用 Wait 办法会被阻塞,直到 计数值 变为0func (wg *WaitGroup) Wait() { // 获取计数、期待数和信号量 statep, semap := wg.state() for { state := atomic.LoadUint64(statep) // 计数值 v := int32(state >> 32) // 期待者数量 w := uint32(state) // 计数值数量为0,间接返回,无需期待 if v == 0 { return } // 到这里阐明计数值数量大于0 // 减少期待者数量:这里会有竞争,比方多个 Wait 调用,或者在同时调用 Add 办法,减少不胜利会持续 for 循环 if atomic.CompareAndSwapUint64(statep, state, state+1) { // 减少胜利后,阻塞在信号量这里,期待被唤醒 runtime_Semacquire(semap) // 被唤醒的时候,应该是0状态。如果重用 WaitGroup,须要等 Wait 返回 if *statep != 0 { panic("sync: WaitGroup is reused before previous Wait has returned") } return } }}
易错点
下面剖析源码能够看到几个会产生 panic
的点,这也是咱们应用 WaitGroup
须要留神的中央
- 计数值变为正数
- 调用 Add 时参数值传正数
func main() {var wg sync.WaitGroupwg.Add(1)wg.Add(-1)wg.Add(-1)}
- 屡次调用 Done 办法
func main() { var wg sync.WaitGroup wg.Add(1) go func() { fmt.Println("test") wg.Done() wg.Done() }() time.Sleep(time.Second) wg.Wait()}
Add 和 Wait 并发调用
Add
和Wait
并发调用,有可能达不到咱们预期的成果,甚至panic
。如下示例中,咱们想要期待 3 个子工作都执行完后再执行主工作,但理论状况可能是子工作还没起来,主工作就持续往下执行了。
func doSomething(wg *sync.WaitGroup) { wg.Add(1) fmt.Println("do something") defer wg.Done()}func main() { var wg sync.WaitGroup for i := 0; i < 3; i++ { go doSomething(&wg) } wg.Wait() fmt.Println("main")}//main//do something//do something
正确的应用形式,应该是在调用 Wait
前先调用 Add
func doSomething(wg *sync.WaitGroup) { defer wg.Done() fmt.Println("do something")}func main() { var wg sync.WaitGroup wg.Add(3) for i := 0; i < 3; i++ { go doSomething(&wg) } wg.Wait() fmt.Println("main")}//do something//do something//do something//main
- 没有等 Wait 返回,就重用 WaitGroup
func main() { var wg sync.WaitGroup wg.Add(1) go func() { fmt.Println("do something") wg.Done() wg.Add(1) }() wg.Wait()}
- 复制应用
咱们晓得 Go 语言中的参数传递,都是值传递,就会产生复制操作。因而在向函数传递 WaitGroup 时,应用指针
进行操作。
// 谬误应用形式,没有应用指针func doSomething(wg sync.WaitGroup) { fmt.Println("do something") defer wg.Done()}func main() { var wg sync.WaitGroup wg.Add(3) for i := 0; i < 3; i++ { // 这里没应用指针,wg状态始终不会扭转,导致 Wait 始终阻塞 go doSomething(wg) } wg.Wait() fmt.Println("main")}
总结
本篇文章,咱们通过源码+示例的形式,一起学习了 sync.WaitGroup
实现逻辑,同时也给出了一些留神点,只有做到如下操作,就不会呈现问题:
- 保障 Add 在 Wait 前调用
- Add 中不传递正数
- 工作实现后不要遗记调用 Done 办法,倡议应用 defer wg.Done()
- 不要复制应用 WaitGroup,函数传递时应用指针传递
- 尽量不复用 WaigGroup,缩小出问题的危险
更多
集体博客: https://lifelmy.github.io/
微信公众号:漫漫Coding路