golang 系列:waitgroup 解析

摘要

Golang 提供了简洁的 go 关键字来让开发者更容易的进行并发编程,同时也提供了 WaitGroup 对象来辅助并发控制。今天我们就来分析下 WaitGroup 的使用方法,顺便瞧一瞧它的底层源码。

WaitGroup 的使用场景和方法

当我们有很多任务要同时进行时,如果并不需要关心各个任务的执行进度,那直接使用 go 关键字即可。

如果我们需要关心所有任务完成后才能往下运行时,则需要 WaitGroup 来阻塞等待这些并发任务了。

WaitGroup 如同它的字面意思,就是等待一组 goroutine 运行完成,主要有三个方法组成:

  • Add(delta int) :添加任务数
  • Wait():阻塞等待所有任务的完成
  • Done():完成任务

下面是它们的具体用法,具体的作用都在注释上:

package main

import (
    "fmt"
    "sync"
    "time"
)

func worker(wg *sync.WaitGroup) {
    doSomething()
    wg.Done() // 2.1、完成任务
}

func main() {
    var wg sync.WaitGroup
    wg.Add(5) // 1、添加 5 个任务
    for i := 1; i <= 5; i++ {
        go worker(&wg) // 2、每个任务并发执行
    }
    wg.Wait() // 3、阻塞等待所有任务完成
}

WaitGroup 源码分析

上面 WaitGroup 的使用很简单,接下来我们到 src/sync/waitgroup.go 里分析下它的源码。首先,是 WaitGroup 的结构体:

type WaitGroup struct {
    noCopy noCopy
    state1 [3]uint32
}

noCopy

其中,noCopy 表示 WaitGroup 是不可复制的。那么什么叫不可复制呢?

举个例子,当我们对函数参数定义了这个不可复制的类型时,开发者只能通过指针来传递函数参数。而规定使用指针传递又有什么好处呢?

好处在于如果有多个函数都定义了这个不可复制的参数时,那么这多个函数参数就可以共用同一个指针变量,来同步执行结果。而 WaitGroup 就是需要这样的约束规定。

state1 字段

接下来我们来看看 WaitGroup 的 state1 字段。state1 是一个包含了 counter 总数、 waiter 等待数、sema 信号量的 uint32 数组。

每当有 goroutine 调用了 Wait() 方法阻塞等待时,就会对 waiter 数量 + 1,然后等待信号量的唤起通知。

当我们调用 Add() 方法时,就会对 state1 的 counter 数量 + 1。

当调用 Done() 方法时就会对 counter 数量 -1。

直到 counter == 0 时就可以通过信号量唤起对应 waiter 数量的 goroutine 了,也就是唤起刚刚阻塞等待的 goroutine 们。

关于信号量的解释,可以参考下 golang 重要知识:mutex 里的相关介绍:

PV 原语解释:
通过操作信号量 S 来处理进程间的同步与互斥的问题。
S>0:表示有 S 个资源可用;S=0 表示无资源可用;S<0 绝对值表示等待队列或链表中的进程个数。信号量 S 的初值应大于等于 0。
P 原语:表示申请一个资源,对 S 原子性的减 1,若 减 1 后仍 S>=0,则该进程继续执行;若 减 1 后 S<0,表示已无资源可用,需要将自己阻塞起来,放到等待队列上。
V 原语:表示释放一个资源,对 S 原子性的加 1;若 加 1 后 S>0,则该进程继续执行;若 加 1 后 S<=0,表示等待队列上有等待进程,需要将第一个等待的进程唤醒。

此处操作系统可以理解为 Go 的运行时 runtime进程可以理解为协程

源码解释

最后,我们来深入 WaitGroup 的三个方法,进行源码分析。大家感兴趣的可以继续往下看,主要是对源码的分析注释。

Add(delta int) 方法

func (wg *WaitGroup) Add(delta int) {
    statep, semap := wg.state()
    if race.Enabled { // 此处是 go 的竞争检测,可以不用关心
        _ = *statep
        if delta < 0 {
            race.ReleaseMerge(unsafe.Pointer(wg))
        }
        race.Disable()
        defer race.Enable()
    }
    state := atomic.AddUint64(statep, uint64(delta)<<32)
    v := int32(state >> 32) // 获取 counter
    w := uint32(state) // 获取 waiter
    if race.Enabled && delta > 0 && v == int32(delta) { // go 的竞争检测,可以不用关心
        race.Read(unsafe.Pointer(semap))
    }
    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 { // counter > 0:还有任务在执行;waiter == 0 表示没有在阻塞等待的 goroutine
        return
    }
    if *statep != state {
        panic("sync: WaitGroup misuse: Add called concurrently with Wait")
    }
    // 执行到此处相当于 countr = 0,即所有的任务都已执行完,需要唤起等待的 goroutine了
    *statep = 0
    for ; w != 0; w-- {
        runtime_Semrelease(semap, false, 0)
    }
}

Done 方法

func (wg *WaitGroup) Done() {
    wg.Add(-1) // 直接调用 Add 方法 对 counter -1
}

Wait 方法

func (wg *WaitGroup) Wait() {
    statep, semap := wg.state()
    if race.Enabled { // go 的竞争检测,可以不用关心
        _ = *statep
        race.Disable()
    }
    for {
        state := atomic.LoadUint64(statep)
        v := int32(state >> 32)
        w := uint32(state)
        if v == 0 {
            // counter 为 0, 不需要再等待了。
            if race.Enabled {
                race.Enable()
                race.Acquire(unsafe.Pointer(wg))
            }
            return
        }
        // waiters 数目 +1.
        if atomic.CompareAndSwapUint64(statep, state, state+1) {
            if race.Enabled && w == 0 {
                race.Write(unsafe.Pointer(semap)) // go 的竞争检测,可以不用关心
            }
            runtime_Semacquire(semap) // 阻塞等待唤起
            if *statep != 0 {
                panic("sync: WaitGroup is reused before previous Wait has returned")
            }
            if race.Enabled {
                race.Enable()
                race.Acquire(unsafe.Pointer(wg))
            }
            return
        }
    }
}

从这几个方法的源码,我们可以看出,Go 并没有使用 mutex 等锁去做字段值修改,而是采用了 atomic 原子操作来进行修改的。这是在底层硬件上支持的,所以性能更好。

总结

WaitGroup 比较简单,就是一些计数值的维护和 goroutine 的阻塞唤起。它的运用也简单,Add、Done、Wait 这三个方法经常是同时出现的。相信大伙深入到源码也能瞧出个大概,这里就献丑了 ㋛。


感兴趣的朋友可以搜一搜公众号「 阅新技术 」,关注更多的推送文章。
可以的话,就顺便点个赞、留个言、分享下,感谢各位支持!
阅新技术,阅读更多的新知识。

©著作权归作者所有,转载或内容合作请联系作者
  • 序言:七十年代末,一起剥皮案震惊了整个滨河市,随后出现的几起案子,更是在滨河造成了极大的恐慌,老刑警刘岩,带你破解...
    沈念sama阅读 195,783评论 5 462
  • 序言:滨河连续发生了三起死亡事件,死亡现场离奇诡异,居然都是意外死亡,警方通过查阅死者的电脑和手机,发现死者居然都...
    沈念sama阅读 82,360评论 2 373
  • 文/潘晓璐 我一进店门,熙熙楼的掌柜王于贵愁眉苦脸地迎上来,“玉大人,你说我怎么就摊上这事。” “怎么了?”我有些...
    开封第一讲书人阅读 142,942评论 0 325
  • 文/不坏的土叔 我叫张陵,是天一观的道长。 经常有香客问我,道长,这世上最难降的妖魔是什么? 我笑而不...
    开封第一讲书人阅读 52,507评论 1 267
  • 正文 为了忘掉前任,我火速办了婚礼,结果婚礼上,老公的妹妹穿的比我还像新娘。我一直安慰自己,他们只是感情好,可当我...
    茶点故事阅读 61,324评论 5 358
  • 文/花漫 我一把揭开白布。 她就那样静静地躺着,像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上,一...
    开封第一讲书人阅读 46,299评论 1 273
  • 那天,我揣着相机与录音,去河边找鬼。 笑死,一个胖子当着我的面吹牛,可吹牛的内容都是我干的。 我是一名探鬼主播,决...
    沈念sama阅读 36,685评论 3 386
  • 文/苍兰香墨 我猛地睁开眼,长吁一口气:“原来是场噩梦啊……” “哼!你这毒妇竟也来了?” 一声冷哼从身侧响起,我...
    开封第一讲书人阅读 35,358评论 0 254
  • 序言:老挝万荣一对情侣失踪,失踪者是张志新(化名)和其女友刘颖,没想到半个月后,有当地人在树林里发现了一具尸体,经...
    沈念sama阅读 39,652评论 1 293
  • 正文 独居荒郊野岭守林人离奇死亡,尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 34,704评论 2 312
  • 正文 我和宋清朗相恋三年,在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 36,465评论 1 326
  • 序言:一个原本活蹦乱跳的男人离奇死亡,死状恐怖,灵堂内的尸体忽然破棺而出,到底是诈尸还是另有隐情,我是刑警宁泽,带...
    沈念sama阅读 32,318评论 3 313
  • 正文 年R本政府宣布,位于F岛的核电站,受9级特大地震影响,放射性物质发生泄漏。R本人自食恶果不足惜,却给世界环境...
    茶点故事阅读 37,711评论 3 299
  • 文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹,春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 28,991评论 0 19
  • 文/苍兰香墨 我抬头看了看天上的太阳。三九已至,却和暖如春,着一层夹袄步出监牢的瞬间,已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 30,265评论 1 251
  • 我被黑心中介骗来泰国打工, 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留,地道东北人。 一个月前我还...
    沈念sama阅读 41,661评论 2 342
  • 正文 我出身青楼,却偏偏与公主长得像,于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子,可洞房花烛夜当晚...
    茶点故事阅读 40,864评论 2 335

推荐阅读更多精彩内容