源码剖析sync.cond(条件变量的实现机制）

ゝ一世哀愁。 2022-10-30 10:27 182阅读 0赞

## 前言 ##

哈喽，大家好，我是`asong`，这是我并发编程系列的第三篇文章，这一篇我们一起来看看`sync.Cond`的使用与实现。之前写过`java`的朋友对等待/通知(wait/notify)机制一定很熟悉，可以利用等待/通知机制实现阻塞或者唤醒，在`Go`语言使用`Cond`也可以达到同样的效果，接下来我们一起来看看它的使用与实现。

## `sync.Cond`的基本使用 ##

`Go`标准库提供了`Cond`原语，为等待/通知场景下的并发问题提供支持。`Cond`他可以让一组的`Goroutine`都在满足特定条件(这个等待条件有很多，可以是某个时间点或者某个变量或一组变量达到了某个阈值，还可以是某个对象的状态满足了特定的条件)时被唤醒，`Cond`是和某个条件相关，这个条件需要一组`goroutine`协作共同完成，在条件还没有满足的时候，所有等待这个条件的`goroutine`都会被阻塞住，只有这一组`goroutine`通过协作达到了这个条件，等待的`goroutine`才可以继续进行下去。

先看这样一个例子：

var (
     done = false
     topic = "Golang梦工厂"
    )
    
    func main() {
     cond := sync.NewCond(&sync.Mutex{})
     go Consumer(topic,cond)
     go Consumer(topic,cond)
     go Consumer(topic,cond)
     Push(topic,cond)
     time.Sleep(5 * time.Second)
    
    }
    
    func Consumer(topic string,cond *sync.Cond)  {
     cond.L.Lock()
     for !done{
      cond.Wait()
     }
     fmt.Println("topic is ",topic," starts Consumer")
     cond.L.Unlock()
    }
    
    func Push(topic string,cond *sync.Cond)  {
     fmt.Println(topic,"starts Push")
     cond.L.Lock()
     done = true
     cond.L.Unlock()
     fmt.Println("topic is ",topic," wakes all")
     cond.Broadcast()
    }
    // 运行结果
    Golang梦工厂 starts Push
    topic is  Golang梦工厂  wakes all
    topic is  Golang梦工厂  starts Consumer
    topic is  Golang梦工厂  starts Consumer
    topic is  Golang梦工厂  starts Consumer

上述代码我们运行了`4`个`Goroutine`，其中三个`Goroutine`分别做了相同的事情，通过调用`cond.Wait()`等特定条件的满足，1个`Goroutine`会调用`cond.Broadcast`唤醒所用陷入等待的`Goroutine`。画个图看一下更清晰：![d248cb3453c2cb86f8aa7307b0485d22.png][]

我们看上面这一段代码，`Cond`使用起来并不简单，使用不当就出现不可避免的问题，所以，有的开发者会认为，`Cond`是唯一难以掌握的`Go`并发原语。为了让大家能更好的理解`Cond`，接下来我们一起看看`Cond`的实现原理。

## `Cond`实现原理 ##

`Cond`的实现还是比较简单的，代码量比较少，复杂的逻辑已经被`Locker`或者`runtime`的等待队列实现了，所以我们来看这些源代码也会轻松一些。首先我们来看一下它的结构体：

type Cond struct {
     noCopy noCopy
    
     // L is held while observing or changing the condition
     L Locker
    
     notify  notifyList
     checker copyChecker
    }

主要有`4`个字段：

*  `nocopy` ：之前在讲`waitGroup`时介绍过，保证结构体不会在编译器期间拷贝，原因就不在这里说了，想了解的看这篇文章[源码剖析sync.WaitGroup(文末思考题你能解释一下吗?)][sync.WaitGroup]
 *  `checker`：用于禁止运行期间发生拷贝，双重检查(`Double check`)
 *  `L`：可以传入一个读写锁或互斥锁，当修改条件或者调用`wait`方法时需要加锁
 *  `notify`：通知链表，调用`wait()`方法的`Goroutine`会放到这个链表中，唤醒从这里取。我们可以看一下`notifyList`的结构：

type notifyList struct {
     wait   uint32
     notify uint32
     lock   uintptr // key field of the mutex
     head   unsafe.Pointer
     tail   unsafe.Pointer
    }

我们简单分析一下`notifyList`的各个字段：

*  `wait`：下一个等待唤醒`Goroutine`的索引，他是在锁外自动递增的.
 *  `notify`：下一个要通知的`Goroutine`的索引，他可以在锁外读取，但是只能在锁持有的情况下写入.
 *  `head`：指向链表的头部
 *  `tail`：指向链表的尾部

基本结构我们都知道了，下面我就来看一看`Cond`提供的三种方法是如何实现的～。

### `wait` ###

我们先来看一下`wait`方法源码部分：

func (c *Cond) Wait() {
     c.checker.check()
     t := runtime_notifyListAdd(&c.notify)
     c.L.Unlock()
     runtime_notifyListWait(&c.notify, t)
     c.L.Lock()
    }

代码量不多，执行步骤如下：

*  执行运行期间拷贝检查，如果发生了拷贝，则直接`panic`程序
 *  调用`runtime_notifyListAdd`将等待计数器加一并解锁；
 *  调用`runtime_notifyListWait`等待其他 `Goroutine` 的唤醒并加锁

`runtime_notifyListAdd`的实现：

// See runtime/sema.go for documentation.
    func notifyListAdd(l *notifyList) uint32 {
     // This may be called concurrently, for example, when called from
     // sync.Cond.Wait while holding a RWMutex in read mode.
     return atomic.Xadd(&l.wait, 1) - 1
    }

代码实现比较简单，原子操作将等待计数器加1，因为`wait`代表的是下一个等待唤醒`Goroutine`的索引，所以需要减1操作。

`runtime_notifyListWait`的实现：

// See runtime/sema.go for documentation.
    func notifyListWait(l *notifyList, t uint32) {
     lockWithRank(&l.lock, lockRankNotifyList)
    
     // Return right away if this ticket has already been notified.
     if less(t, l.notify) {
      unlock(&l.lock)
      return
     }
    
     // Enqueue itself.
     s := acquireSudog()
     s.g = getg()
     s.ticket = t
     s.releasetime = 0
     t0 := int64(0)
     if blockprofilerate > 0 {
      t0 = cputicks()
      s.releasetime = -1
     }
     if l.tail == nil {
      l.head = s
     } else {
      l.tail.next = s
     }
     l.tail = s
     goparkunlock(&l.lock, waitReasonSyncCondWait, traceEvGoBlockCond, 3)
     if t0 != 0 {
      blockevent(s.releasetime-t0, 2)
     }
     releaseSudog(s)
    }

这里主要执行步骤如下：

*  检查当前`wait`与`notify`索引位置是否匹配，如果已经被通知了，便立即返回.
 *  获取当前`Goroutine`，并将当前`Goroutine`追加到链表末端.
 *  调用`goparkunlock`方法让当前`Goroutine`进入等待状态，也就是进入睡眠，等待唤醒
 *  被唤醒后，调用`releaseSudog`释放当前等待列表中的`Goroutine`

看完源码我们来总结一下注意事项：

`wait`方法会把调用者放入`Cond`的等待队列中并阻塞，直到被唤醒，调用`wait`方法必须要持有`c.L`锁。

### `signal`和`Broadcast` ###

`signal`和`Broadcast`都会唤醒等待队列，不过`signal`是唤醒链表最前面的`Goroutine`，`Boradcast`会唤醒队列中全部的`Goroutine`。下面我们分别来看一下`signal`和`broadcast`的源码：

*  `signal`

func (c *Cond) Signal() {
     c.checker.check()
     runtime_notifyListNotifyOne(&c.notify)
    }
    func notifyListNotifyOne(l *notifyList) {
     if atomic.Load(&l.wait) == atomic.Load(&l.notify) {
      return
     }
     lockWithRank(&l.lock, lockRankNotifyList)
     t := l.notify
     if t == atomic.Load(&l.wait) {
      unlock(&l.lock)
      return
     }
    
     atomic.Store(&l.notify, t+1)
    
     for p, s := (*sudog)(nil), l.head; s != nil; p, s = s, s.next {
      if s.ticket == t {
       n := s.next
       if p != nil {
        p.next = n
       } else {
        l.head = n
       }
       if n == nil {
        l.tail = p
       }
       unlock(&l.lock)
       s.next = nil
       readyWithTime(s, 4)
       return
      }
     }
     unlock(&l.lock)
    }

上面我们看`wait`源代码时，每次都会调用都会原子递增`wait`，那么这个`wait`就代表当前最大的`wait`值，对应唤醒的时候，也就会对应一个`notify`属性，我们在`notifyList`链表中逐个检查，找到`ticket`对应相等的`notify`属性。这里大家肯定会有疑惑，我们为何不直接取链表头部唤醒呢？

`notifyList`并不是一直有序的，`wait`方法中调用`runtime_notifyListAdd`和`runtime_notifyListWait`完全是两个独立的行为，中间还有释放锁的行为，而当多个 `goroutine` 同时进行时，中间会产生进行并发操作，这样就会出现乱序，所以采用这种操作即使在 `notifyList` 乱序的情况下，也能取到最先`Wait` 的 `goroutine`。

*  `broadcast`

func (c *Cond) Broadcast() {
     c.checker.check()
     runtime_notifyListNotifyAll(&c.notify)
    }
    func notifyListNotifyAll(l *notifyList) {
     if atomic.Load(&l.wait) == atomic.Load(&l.notify) {
      return
     }
    
     lockWithRank(&l.lock, lockRankNotifyList)
     s := l.head
     l.head = nil
     l.tail = nil
    
     atomic.Store(&l.notify, atomic.Load(&l.wait))
     unlock(&l.lock)
     for s != nil {
      next := s.next
      s.next = nil
      readyWithTime(s, 4)
      s = next
     }
    }

全部唤醒实现要简单一些，主要是通过调用`readyWithTime`方法唤醒链表中的`goroutine`，唤醒顺序也是按照加入队列的先后顺序，先加入的会先被唤醒，而后加入的可能 `Goroutine` 需要等待调度器的调度。

最后我们总结一下使用这两个方法要注意的问题：

*  `Signal`：允许调用者唤醒一个等待此`Cond`的`Goroutine`，如果此时没有等待的 `goroutine`，显然无需通知`waiter`；如果`Cond` 等待队列中有一个或者多个等待的 `goroutine`，则需要从等待队列中移除第一个 `goroutine` 并把它唤醒。调用 `Signal`方法时，不强求你一定要持有 `c.L` 的锁。
 *  `broadcast`：允许调用者唤醒所有等待此 `Cond` 的`goroutine`。如果此时没有等待的`goroutine`，显然无需通知 waiter；如果 `Cond` 等待队列中有一个或者多个等待的 `goroutine`，则清空所有等待的 `goroutine`，并全部唤醒，不强求你一定要持有 `c.L` 的锁。

## 注意事项 ##

*  调用`wait`方法的时候一定要加锁，否则会导致程序发生`panic`.
 *  `wait`调用时需要检查等待条件是否满足，也就说`goroutine`被唤醒了不等于等待条件被满足，等待者被唤醒，只是得到了一次检查的机会而已，推荐写法如下：

//    c.L.Lock()
    //    for !condition() {
    //        c.Wait()
    //    }
    //    ... make use of condition ...
    //    c.L.Unlock()

*  `Signal` 和 `Boardcast` 两个唤醒操作不需要加锁

## 总结 ##

其实`Cond`在实际项目中被使用的机会比较少，`Go`特有的`channel`就可以代替它，暂时只在`Kubernetes`项目中看到了应用，使用场景是每次往队列中成功增加了元素后就需要调用 `Broadcast` 通知所有的等待者，使用`Cond`就很合适，相比`channel`减少了代码复杂性。

**好啦，这篇文章就到这里啦，素质三连（分享、点赞、在看）都是笔者持续创作更多优质内容的动力！**

**创建了一个Golang学习交流群，欢迎各位大佬们踊跃入群，我们一起学习交流。入群方式：加我vx拉你入群，或者公众号获取入群二维码**

**结尾给大家发一个小福利吧，最近我在看\[微服务架构设计模式\]这一本书，讲的很好，自己也收集了一本PDF，有需要的小伙可以到自行下载。获取方式：关注公众号：\[Golang梦工厂\]，后台回复：\[微服务\]，即可获取。**

**我翻译了一份GIN中文文档，会定期进行维护，有需要的小伙伴后台回复\[gin\]即可下载。**

**翻译了一份Machinery中文文档，会定期进行维护，有需要的小伙伴们后台回复\[machinery\]即可获取。**

**我是asong，一名普普通通的程序猿，让我们一起慢慢变强吧。欢迎各位的关注，我们下期见~~~**

![580623cf892e9d27a2f85ac1f9645390.png][]

推荐往期文章：

*  [Go看源码必会知识之unsafe包][Go_unsafe]
 *  [源码剖析panic与recover，看不懂你打我好了！][panic_recover]
 *  [详解并发编程基础之原子操作(atomic包)][atomic]
 *  [详解defer实现机制][defer]
 *  [真的理解interface了嘛][interface]
 *  [Leaf—Segment分布式ID生成系统（Golang实现版本）][Leaf_Segment_ID_Golang]
 *  [十张动图带你搞懂排序算法(附go实现代码)][go]
 *  [go参数传递类型][go 1]
 *  [手把手教姐姐写消息队列][Link 1]
 *  [常见面试题之缓存雪崩、缓存穿透、缓存击穿][Link 2]
 *  [详解Context包，看这一篇就够了！！！][Context]
 *  [go-ElasticSearch入门看这一篇就够了(一)][go-ElasticSearch]
 *  [面试官：go中for-range使用过吗？这几个问题你能解释一下原因吗][go_for-range]

[d248cb3453c2cb86f8aa7307b0485d22.png]: /images/20221024/d533dbdaaf3d4881b1476660d528fecd.png
[sync.WaitGroup]: https://mp.weixin.qq.com/s?__biz=MzIzMDU0MTA3Nw%3D%3D&idx=1&mid=2247484551&scene=21&sn=aff008de8b6c089d7daf9c963ce8b60a#wechat_redirect
[580623cf892e9d27a2f85ac1f9645390.png]: /images/20221024/ce098db289ab4776813f91cf192335e6.png
[Go_unsafe]: https://mp.weixin.qq.com/s?__biz=MzIzMDU0MTA3Nw%3D%3D&idx=1&mid=2247484529&scene=21&sn=8653947ab70f4d73ff6260965be32460#wechat_redirect
[panic_recover]: https://mp.weixin.qq.com/s?__biz=MzIzMDU0MTA3Nw%3D%3D&idx=1&mid=2247484460&scene=21&sn=0838818da3a456f409f70dd321ffa2e6#wechat_redirect
[atomic]: https://mp.weixin.qq.com/s?__biz=MzIzMDU0MTA3Nw%3D%3D&idx=1&mid=2247484475&scene=21&sn=f470d42fee8b9b65fad7b43e57518cfc#wechat_redirect
[defer]: https://mp.weixin.qq.com/s?__biz=MzIzMDU0MTA3Nw%3D%3D&idx=1&mid=2247484410&scene=21&sn=f3bf9d49c5a746b217a4d404d2a573f3#wechat_redirect
[interface]: https://mp.weixin.qq.com/s?__biz=MzIzMDU0MTA3Nw%3D%3D&idx=1&mid=2247484349&scene=21&sn=edfd2b27a8331f1f52e9f0f74391bd8e#wechat_redirect
[Leaf_Segment_ID_Golang]: https://mp.weixin.qq.com/s?__biz=MzIzMDU0MTA3Nw%3D%3D&idx=1&mid=2247484313&scene=21&sn=6859d5eec70a1b3844f3bd2fc430742c#wechat_redirect
[go]: https://mp.weixin.qq.com/s?__biz=MzIzMDU0MTA3Nw%3D%3D&idx=1&mid=2247484201&scene=21&sn=7fda75e8da287bd3bbe5a719d9e72688#wechat_redirect
[go 1]: https://mp.weixin.qq.com/s?__biz=MzIzMDU0MTA3Nw%3D%3D&idx=1&mid=2247484131&scene=21&sn=5b1608613f6f1568acc85214e0ff1cbd#wechat_redirect
[Link 1]: https://mp.weixin.qq.com/s?__biz=MzIzMDU0MTA3Nw%3D%3D&idx=1&mid=2247483933&scene=21&sn=45c609694ef8cd34c98f5bd74903c754#wechat_redirect
[Link 2]: https://mp.weixin.qq.com/s?__biz=MzIzMDU0MTA3Nw%3D%3D&chksm=e8b0902edfc71938f7d7a29246d7278ac48e6c104ba27c684e12e840892252b0823de94b94c1&idx=1&lang=zh_CN&mid=2247483988&scene=21&sn=3bd52650907867d65f1c4d5c3cff8f13&token=1558933779#wechat_redirect
[Context]: https://mp.weixin.qq.com/s?__biz=MzIzMDU0MTA3Nw%3D%3D&idx=1&mid=2247483915&scene=21&sn=d2df74f983fcaedc50db1962d4d3e873#wechat_redirect
[go-ElasticSearch]: https://mp.weixin.qq.com/s?__biz=MzIzMDU0MTA3Nw%3D%3D&idx=1&mid=2247483889&scene=21&sn=6295c2cead69521b36287a46ca9175ea#wechat_redirect
[go_for-range]: https://mp.weixin.qq.com/s?__biz=MzIzMDU0MTA3Nw%3D%3D&idx=1&mid=2247483875&scene=21&sn=c5da523333e807dddb70228abc2a05b9#wechat_redirect