• Go sync.WaitGroup的学习


    一. 前言

    WaitGroup是Golang应用开发过程中经常使用的并发控制技术。

    WaitGroup,可理解为Wait-Goroutine-Group,即等待一组goroutine结束。比如某个goroutine需要等待其他几个goroutine全部完成,那么使用WaitGroup可以轻松实现。

    下面程序展示了一个goroutine等待另外两个goroutine结束的例子:

    package main
    
    import (
        "fmt"
        "time"
        "sync"
    )
    
    func main() {
        var wg sync.WaitGroup
    
        wg.Add(2) //设置计数器,数值即为goroutine的个数
        go func() {
            //Do some work
            time.Sleep(1*time.Second)
    
            fmt.Println("Goroutine 1 finished!")
            wg.Done() //goroutine执行结束后将计数器减1
        }()
    
        go func() {
            //Do some work
            time.Sleep(2*time.Second)
    
            fmt.Println("Goroutine 2 finished!")
            wg.Done() //goroutine执行结束后将计数器减1
        }()
    
        wg.Wait() //主goroutine阻塞等待计数器变为0
        fmt.Printf("All Goroutine finished!")
    }
    
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31

    简单的说,上面程序中wg内部维护了一个计数器:

    1. 启动goroutine前将计数器通过Add(2)将计数器设置为待启动的goroutine个数。
    2. 启动goroutine后,使用Wait()方法阻塞自己,等待计数器变为0。
    3. 每个goroutine执行结束通过Done()方法将计数器减1。
    4. 计数器变为0后,阻塞的goroutine被唤醒。

    其实WaitGroup也可以实现一组goroutine等待另一组goroutine,这有点像玩杂技,很容出错,如果不了解其实现原理更是如此。实际上,WaitGroup的实现源码非常简单。

    二. 夯实基础

    2.1 信号量

    信号量是Unix系统提供的一种保护共享资源的机制,用于防止多个线程同时访问某个资源。
    可简单理解为信号量为一个数值:

    • 当信号量>0时,表示资源可用,获取信号量时系统自动将信号量减1;
    • 当信号量==0时,表示资源暂不可用,获取信号量时,当前线程会进入睡眠,当信号量为正时被唤醒;

    由于WaitGroup实现中也使用了信号量,在此做个简单介绍。

    2.2 sync.WaitGroup

    源码包中src/sync/waitgroup.go:WaitGroup定义了其数据结构:

    type WaitGroup struct {
        state1 [3]uint32
    }
    
    • 1
    • 2
    • 3

    state1是个长度为3的数组,其中包含了state和一个信号量,而state实际上是两个计数器:

    • counter: 当前还未执行结束的goroutine计数器
    • waiter count: 等待goroutine-group结束的goroutine数量,即有多少个等候者
    • semaphore: 信号量

    考虑到字节是否对齐,三者出现的位置不同,为简单起见,依照字节已对齐情况下,三者在内存中的位置如下所示:在这里插入图片描述
    WaitGroup对外提供三个接口:

    1. Add(delta int): 将delta值加到counter中
    2. Wait(): waiter递增1,并阻塞等待信号量semaphore
    3. Done(): counter递减1,按照waiter数值释放相应次数信号量

    等待组有下面几个方法可用,如下表所示。
    在这里插入图片描述
    下面分别介绍这三个函数的实现细节。

    2.2.1 Add(delta int)

    Add()做了两件事,一是把delta值累加到counter中,因为delta可以为负值,也就是说counter有可能变成0或负值,所以第二件事就是当counter值变为0时,根据waiter数值释放等量的信号量,把等待的goroutine全部唤醒,如果counter变为负值,则panic.

    Add()伪代码如下:

    func (wg *WaitGroup) Add(delta int) {
        statep, semap := wg.state() //获取state和semaphore地址指针
    
        state := atomic.AddUint64(statep, uint64(delta)<<32) //把delta左移32位累加到state,即累加到counter中
        v := int32(state >> 32) //获取counter值
        w := uint32(state)      //获取waiter值
    
        if v < 0 {              //经过累加后counter值变为负值,panic
            panic("sync: negative WaitGroup counter")
        }
    
        //经过累加后,此时,counter >= 0
        //如果counter为正,说明不需要释放信号量,直接退出
        //如果waiter为零,说明没有等待者,也不需要释放信号量,直接退出
        if v > 0 || w == 0 {
            return
        }
    
        //此时,counter一定等于0,而waiter一定大于0(内部维护waiter,不会出现小于0的情况),
        //先把counter置为0,再释放waiter个数的信号量
        *statep = 0
        for ; w != 0; w-- {
            runtime_Semrelease(semap, false) //释放信号量,执行一次释放一个,唤醒一个等待者
        }
    }
    
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25

    2.2.2 Wait()

    Wait()方法也做了两件事,一是累加waiter, 二是阻塞等待信号量

    func (wg *WaitGroup) Wait() {
        statep, semap := wg.state() //获取state和semaphore地址指针
        for {
            state := atomic.LoadUint64(statep) //获取state值
            v := int32(state >> 32)            //获取counter值
            w := uint32(state)                 //获取waiter值
            if v == 0 {                        //如果counter值为0,说明所有goroutine都退出了,不需要待待,直接返回
                return
            }
    
            // 使用CAS(比较交换算法)累加waiter,累加可能会失败,失败后通过for loop下次重试
            if atomic.CompareAndSwapUint64(statep, state, state+1) {
                runtime_Semacquire(semap) //累加成功后,等待信号量唤醒自己
                return
            }
        }
    }
    
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17

    这里用到了CAS算法保证有多个goroutine同时执行Wait()时也能正确累加waiter。

    2.2.3 Done()

    Done()只做一件事,即把counter减1,我们知道Add()可以接受负值,所以Done实际上只是调用了Add(-1)。
    源码如下:

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

    Done()的执行逻辑就转到了Add(),实际上也正是最后一个完成的goroutine把等待者唤醒的。

    2.3 sync.WaitGroup的使用

    sync.WaitGroup类型(以下简称WaitGroup类型)是开箱即用的,也是并发安全的。

    WaitGroup类型拥有三个指针方法:Add、Done和Wait。你可以想象该类型中有一个计数器,它的默认值是0。我们可以通过调用该类型值的Add方法来增加,或者减少这个计数器的值。

    一般情况下,我会用这个方法来记录需要等待的 goroutine 的数量。相对应的,这个类型的Done方法,用于对其所属值中计数器的值进行减一操作。我们可以在需要等待的 goroutine 中,通过defer语句调用它。

    而此类型的Wait方法的功能是,阻塞当前的 goroutine,直到其所属值中的计数器归零。如果在该方法被调用的时候,那个计数器的值就是0,那么它将不会做任何事情。

    三. 应用实践

    3.1 sync.WaitGroup类型值中计数器的值可以小于0吗?

    这里的典型回答是:不可以。

    问题解析

    为什么不可以呢,我们解析一下。之所以说WaitGroup值中计数器的值不能小于0,是因为这样会引发一个 panic。 不适当地调用这类值的Done方法和Add方法都会如此。别忘了,我们在调用Add方法的时候是可以传入一个负数的。

    实际上,导致WaitGroup值的方法抛出 panic 的原因不只这一种。

    你需要知道,在我们声明了这样一个变量之后,应该首先根据需要等待的 goroutine,或者其他事件的数量,调用它的Add方法,以使计数器的值大于0。这是确保我们能在后面正常地使用这类值的前提。

    如果我们对它的Add方法的首次调用,与对它的Wait方法的调用是同时发起的,比如,在同时启用的两个 goroutine 中,分别调用这两个方法,那么就有可能会让这里的Add方法抛出一个 panic。

    这种情况不太容易复现,也正因为如此,我们更应该予以重视。所以,虽然WaitGroup值本身并不需要初始化,但是尽早地增加其计数器的值,还是非常有必要的。

    另外,你可能已经知道,WaitGroup值是可以被复用的,但需要保证其计数周期的完整性。这里的计数周期指的是这样一个过程:该值中的计数器值由0变为了某个正整数,而后又经过一系列的变化,最终由某个正整数又变回了0。

    也就是说,只要计数器的值始于0又归为0,就可以被视为一个计数周期。在一个此类值的生命周期中,它可以经历任意多个计数周期。但是,只有在它走完当前的计数周期之后,才能够开始下一个计数周期。在这里插入图片描述
    因此,也可以说,如果一个此类值的Wait方法在它的某个计数周期中被调用,那么就会立即阻塞当前的 goroutine,直至这个计数周期完成。在这种情况下,该值的下一个计数周期,必须要等到这个Wait方法执行结束之后,才能够开始。

    如果在一个此类值的Wait方法被执行期间,跨越了两个计数周期,那么就会引发一个 panic

    例如,在当前的 goroutine 因调用此类值的Wait方法,而被阻塞的时候,另一个 goroutine 调用了该值的Done方法,并使其计数器的值变为了0。

    这会唤醒当前的 goroutine,并使它试图继续执行Wait方法中其余的代码。但在这时,又有一个 goroutine 调用了它的Add方法,并让其计数器的值又从0变为了某个正整数。此时,这里的Wait方法就会立即抛出一个 panic。

    纵观上述会引发 panic 的后两种情况,我们可以总结出这样一条关于WaitGroup值的使用禁忌,即:不要把增加其计数器值的操作和调用其Wait方法的代码,放在不同的 goroutine 中执行。换句话说,要杜绝对同一个WaitGroup值的两种操作的并发执行。

    除了第一种情况外,我们通常需要反复地实验,才能够让WaitGroup值的方法抛出 panic。再次强调,虽然这不是每次都发生,但是在长期运行的程序中,这种情况发生的概率还是不小的,我们必须要重视它们。

    如果你对复现这些异常情况感兴趣,那么可以参看sync代码包中的 waitgroup_test.go 文件。其中的名称以TestWaitGroupMisuse为前缀的测试函数,很好地展示了这些异常情况的发生条件。

    四. 源码学习

    五. 总结

    5.1 总结概述

    简单说来,WaitGroup通常用于等待一组“工作协程”结束的场景,其内部维护两个计数器,这里把它们称为“工作协程”计数器和“坐等协程”计数器,WaitGroup对外提供的三个方法分工非常明确

    1. Add(delta int)方法用于增加“工作协程”计数,通常在启动新的“工作协程”之前调用;
    2. Done()方法用于减少“工作协程”计数,每次调用递减1,通常在“工作协程”内部且在临近返回之前调用;
    3. Wait()方法用于增加“坐等协程”计数,通常在所有”工作协程”全部启动之后调用;

    Done()方法除了负责递减“工作协程”计数以外,还会在“工作协程”计数变为0时检查“坐等协程”计数器并把“坐等协程”唤醒。

    需要注意的是,Done()方法递减“工作协程”计数后,如果“工作协程”计数变成负数时,将会触发panic,这就要求Add()方法调用要早于Done()方法。

    此外,通过Add()方法累加的“工作协程”计数要与实际需要等待的“工作协程”数量一致,否则也会触发panic。

    • 当“工作协程”计数多于实际需要等待的“工作协程”数量时,“坐等协程”可能会永远无法被唤醒而产生列锁,此时,Go运行时检测到死锁会触发panic,
    • 当“工作协程”计数小于实际需要等待的“工作协程”数量时,Done()会在“工作协程”计数变为负数时触发panic。

    5.2 高效编码技巧

    1. 最好用“先统一Add,再并发Done,最后Wait”这种标准方式,来使用WaitGroup值

    参考资料

    1. waitGroup—《Go专家编程》
  • 相关阅读:
    541、RabbitMQ详细入门教程系列 -【Jackson2JsonMessageConvert】 2022.09.05
    Kotlin 学习笔记(七)operator约定
    技术干货 | 一文弄懂差分隐私原理!
    《微信小程序-进阶篇》Lin-ui组件库源码分析-列表组件List(三)
    基于Openwrt系统架构,实现应用与驱动的实例。
    希望科怀早日康复
    Centos安装RabbitMQ超详细(必须收藏)
    洗地机哪个牌子好?2023热门洗地推荐
    Jvm之内存泄漏
    重装系统 + C盘怎么设置 (如何设置将电脑软件安装在非c 盘)
  • 原文地址:https://blog.csdn.net/qq_41893274/article/details/128055029