WaitGroup原理分析

2023-10-11 23:12
文章标签 分析 原理 waitgroup

本文主要是介绍WaitGroup原理分析,希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!

背景

在实际业务开发中,我们会遇到以下场景:请求数据库,批量获取1000条数据记录后,处理数据
为了减少因一次批量获取的数据太多,导致的数据库延时增加,我们可以把一次请求拆分成多次请求,并发去处理,当所有的并发请求完成后,再继续处理这些返回的数据
golang中的WaitGroup,就可以帮助我们实现上述的场景

快速入门

背景:开启10个goroutine并发执行,等待所有goroutine执行完成后,当前goroutine打印执行完成

func TestWaitGroup(t *testing.T) {var wg sync.WaitGroupfor i := 0; i < 10; i++ {index := igo func() {wg.Add(1)defer wg.Done()fmt.Println(fmt.Sprintf("%+v 正在执行", index))}()}wg.Wait()fmt.Println("TestWaitGroup method done")
}

源码分析

golang版本:1.18.2

源码路径:src/sync/waitgroup.go

// A WaitGroup waits for a collection of goroutines to finish.
// The main goroutine calls Add to set the number of
// goroutines to wait for. Then each of the goroutines
// runs and calls Done when finished. At the same time,
// Wait can be used to block until all goroutines have finished.
//
// A WaitGroup must not be copied after first use.
// WaitGroup 等待 goroutine 集合完成
// 主 goroutine 调用 Add 设置等待的 goroutine 数量
// 然后每个 goroutine 运行并在完成时调用 Done
// 同时,Wait 可以用来阻塞,直到所有 goroutine 都完成
type WaitGroup struct {noCopy noCopy// 64-bit value: high 32 bits are counter, low 32 bits are waiter count.// 64-bit atomic operations require 64-bit alignment, but 32-bit// compilers only guarantee that 64-bit fields are 32-bit aligned.// For this reason on 32 bit architectures we need to check in state()// if state1 is aligned or not, and dynamically "swap" the field order if// needed.// 64位值:高32位是计数器,低32位是waiter计数// 64位原子操作需要64位对齐,但32位编译器仅保证64位字段是32位对齐的// 因此,在 32 位架构上,我们需要在 state() 中检查 state1 是否对齐,并在需要时动态“交换”字段顺序state1 uint64state2 uint32
}

noCopy:WaitGroup在首次使用后,不能被复制
state1,state2:一共占用12字节,保存了三类信息:4字节保存goroutine计数,4字节保存waiter计数,4字节保存信号量
WaitGroup对外提供了以下三个方法:

// 设置等待的goroutine数量
func (wg *WaitGroup) Add(delta int)
// goroutine执行完成
func (wg *WaitGroup) Done()
// 阻塞等待所有的goroutine都执行完成
func (wg *WaitGroup) Wait()

Add

// state returns pointers to the state and sema fields stored within wg.state*.
func (wg *WaitGroup) state() (statep *uint64, semap *uint32) {if unsafe.Alignof(wg.state1) == 8 || uintptr(unsafe.Pointer(&wg.state1))%8 == 0 {// state1 is 64-bit aligned: nothing to do.return &wg.state1, &wg.state2} else {// state1 is 32-bit aligned but not 64-bit aligned: this means that// (&state1)+4 is 64-bit aligned.state := (*[3]uint32)(unsafe.Pointer(&wg.state1))return (*uint64)(unsafe.Pointer(&state[1])), &state[0]}
}// Add adds delta, which may be negative, to the WaitGroup counter.
// If the counter becomes zero, all goroutines blocked on Wait are released.
// If the counter goes negative, Add panics.
//
// Note that calls with a positive delta that occur when the counter is zero
// must happen before a Wait. Calls with a negative delta, or calls with a
// positive delta that start when the counter is greater than zero, may happen
// at any time.
// Typically this means the calls to Add should execute before the statement
// creating the goroutine or other event to be waited for.
// If a WaitGroup is reused to wait for several independent sets of events,
// new Add calls must happen after all previous Wait calls have returned.
// See the WaitGroup example.
// Add 将 delta(可能为负)添加到 WaitGroup 计数器。
// 如果计数器变为零,则所有在 Wait 上阻塞的 goroutine 都会被释放。
// 如果计数器变为负数,则添加panic。
// 请注意,计数器为零时发生的具有正增量的调用必须在等待之前发生。 
// 具有负增量的调用或在计数器大于零时开始的具有正增量的调用可能随时发生。
// 通常,这意味着对 Add 的调用应该在创建 goroutine 或其他要等待的事件的语句之前执行。
// 如果重用一个 WaitGroup 来等待几个独立的事件集,新的 Add 调用必须在所有先前的 Wait 调用返回后发生。
func (wg *WaitGroup) Add(delta int) {statep, semap := wg.state()if race.Enabled {_ = *statep // trigger nil deref earlyif delta < 0 {// Synchronize decrements with Wait.race.ReleaseMerge(unsafe.Pointer(wg))}race.Disable()defer race.Enable()}// 记录goroutine计数state := atomic.AddUint64(statep, uint64(delta)<<32)// 获取goroutine计数v := int32(state >> 32)// 获取waiter计数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(semap))}// goroutine计数小于0if v < 0 {panic("sync: negative WaitGroup counter")}// w != 0说明已经执行了Wait且还有阻塞等待的goroutine,此时不允许在执行Addif w != 0 && delta > 0 && v == int32(delta) {panic("sync: WaitGroup misuse: Add called concurrently with Wait")}// 存在没有执行完成的goroutine,或者当前没有waiter,直接返回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.// 此时goroutine计数为0,且waiter计数大于0,不然上一步就返回了// 现在以下状态不能同时发生:// 1. 并发调用Add和Wait// 2. 当goroutine计数为0时,Wait不会继续增加waiter计数// 仍然做一个廉价的健全性检查来检测 WaitGroup 的滥用,防止以上情况发生if *statep != state {panic("sync: WaitGroup misuse: Add called concurrently with Wait")}// Reset waiters count to 0.// 重置waiter计数*statep = 0// 唤醒所有的waiterfor ; w != 0; w-- {runtime_Semrelease(semap, false, 0)}
}

delta代表本次需要记录的goroutine计数,可能为负数
64位原子操作需要64位对齐,但32位编译器仅保证64位字段是32位对齐的
当state1是64位对齐时,state1高32位是goroutine计数,低32位是waiter计数
当state1不是64位对齐时,动态“交换”字段顺序
记录goroutine计数的变化delta
如果goroutine计数小于0,则直接panic
如果已经执行了Wait且还有阻塞等待的goroutine,此时不允许在执行Add
如果存在没有执行完成的goroutine,或者当前没有waiter,直接返回
当goroutine计数为0,且waiter计数大于0时,现在以下状态不能同时发生:

并发调用Add和Wait
当goroutine计数为0时,Wait不会继续增加waiter计数

简单校验通过后,重置waiter计数为0,唤醒所有阻塞等待的waiter

Done

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

调用Add,delta = -1,代表goroutine计数-1

Wait

// Wait blocks until the WaitGroup counter is zero.
func (wg *WaitGroup) Wait() {statep, semap := wg.state()if race.Enabled {_ = *statep // trigger nil deref earlyrace.Disable()}for {state := atomic.LoadUint64(statep)// 获取goroutine计数v := int32(state >> 32)// 获取waiter计数w := uint32(state)// goroutine计数为0,不需要等待,直接返回if v == 0 {// Counter is 0, no need to wait.if race.Enabled {race.Enable()race.Acquire(unsafe.Pointer(wg))}return}// Increment waiters count.// waiter计数+1if atomic.CompareAndSwapUint64(statep, 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(semap))}// 阻塞,等待goroutine计数为0后唤醒继续执行runtime_Semacquire(semap)// Wait还没有执行完成,就开始复用WaitGroupif *statep != 0 {panic("sync: WaitGroup is reused before previous Wait has returned")}if race.Enabled {race.Enable()race.Acquire(unsafe.Pointer(wg))}return}}
}

调用state(),保证字段内存对齐
如果goroutine计数为0,不需要等待,直接返回
尝试对waiter计数+1,若失败,则继续下一轮重试
对waiter计数+1成功,则阻塞当前goroutine,等待goroutine计数为0后唤醒继续执行
唤醒继续执行后,简单判断是否存在Wait还没有执行完成,就开始复用WaitGroup的情况,如果有,则panic;如果没有,则直接返回

这篇关于WaitGroup原理分析的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!



http://www.chinasem.cn/article/191533

相关文章

ShardingProxy读写分离之原理、配置与实践过程

《ShardingProxy读写分离之原理、配置与实践过程》ShardingProxy是ApacheShardingSphere的数据库中间件,通过三层架构实现读写分离,解决高并发场景下数据库性能瓶... 目录一、ShardingProxy技术定位与读写分离核心价值1.1 技术定位1.2 读写分离核心价值二

深度解析Python中递归下降解析器的原理与实现

《深度解析Python中递归下降解析器的原理与实现》在编译器设计、配置文件处理和数据转换领域,递归下降解析器是最常用且最直观的解析技术,本文将详细介绍递归下降解析器的原理与实现,感兴趣的小伙伴可以跟随... 目录引言:解析器的核心价值一、递归下降解析器基础1.1 核心概念解析1.2 基本架构二、简单算术表达

深入浅出Spring中的@Autowired自动注入的工作原理及实践应用

《深入浅出Spring中的@Autowired自动注入的工作原理及实践应用》在Spring框架的学习旅程中,@Autowired无疑是一个高频出现却又让初学者头疼的注解,它看似简单,却蕴含着Sprin... 目录深入浅出Spring中的@Autowired:自动注入的奥秘什么是依赖注入?@Autowired

从原理到实战解析Java Stream 的并行流性能优化

《从原理到实战解析JavaStream的并行流性能优化》本文给大家介绍JavaStream的并行流性能优化:从原理到实战的全攻略,本文通过实例代码给大家介绍的非常详细,对大家的学习或工作具有一定的... 目录一、并行流的核心原理与适用场景二、性能优化的核心策略1. 合理设置并行度:打破默认阈值2. 避免装箱

Python中的filter() 函数的工作原理及应用技巧

《Python中的filter()函数的工作原理及应用技巧》Python的filter()函数用于筛选序列元素,返回迭代器,适合函数式编程,相比列表推导式,内存更优,尤其适用于大数据集,结合lamb... 目录前言一、基本概念基本语法二、使用方式1. 使用 lambda 函数2. 使用普通函数3. 使用 N

MyBatis-Plus 与 Spring Boot 集成原理实战示例

《MyBatis-Plus与SpringBoot集成原理实战示例》MyBatis-Plus通过自动配置与核心组件集成SpringBoot实现零配置,提供分页、逻辑删除等插件化功能,增强MyBa... 目录 一、MyBATis-Plus 简介 二、集成方式(Spring Boot)1. 引入依赖 三、核心机制

redis和redission分布式锁原理及区别说明

《redis和redission分布式锁原理及区别说明》文章对比了synchronized、乐观锁、Redis分布式锁及Redission锁的原理与区别,指出在集群环境下synchronized失效,... 目录Redis和redission分布式锁原理及区别1、有的同伴想到了synchronized关键字

Android 缓存日志Logcat导出与分析最佳实践

《Android缓存日志Logcat导出与分析最佳实践》本文全面介绍AndroidLogcat缓存日志的导出与分析方法,涵盖按进程、缓冲区类型及日志级别过滤,自动化工具使用,常见问题解决方案和最佳实... 目录android 缓存日志(Logcat)导出与分析全攻略为什么要导出缓存日志?按需过滤导出1. 按

Linux中的HTTPS协议原理分析

《Linux中的HTTPS协议原理分析》文章解释了HTTPS的必要性:HTTP明文传输易被篡改和劫持,HTTPS通过非对称加密协商对称密钥、CA证书认证和混合加密机制,有效防范中间人攻击,保障通信安全... 目录一、什么是加密和解密?二、为什么需要加密?三、常见的加密方式3.1 对称加密3.2非对称加密四、

MySQL中读写分离方案对比分析与选型建议

《MySQL中读写分离方案对比分析与选型建议》MySQL读写分离是提升数据库可用性和性能的常见手段,本文将围绕现实生产环境中常见的几种读写分离模式进行系统对比,希望对大家有所帮助... 目录一、问题背景介绍二、多种解决方案对比2.1 原生mysql主从复制2.2 Proxy层中间件:ProxySQL2.3