Go语言sync.Cond基本使用及原理示例详解
1. 简介
2. 基本使用
2.1 定义
2.2 方法说明
2.3 使用方式
2.4 使用例子
2.5 为什么Sync.Cond 需要关联一个锁，然后调用Wait方法前需要先获取该锁
3.使用场景
3.1 基本说明
3.2 场景说明
3.2.1 同步和协调多个协程之间共享资源
3.2.2 需要重复唤醒的场景中使用
4. 原理
4.1 基本原理
4.2 实现
4.2.1 Wait方法实现
4.2.2 Singal方法实现
4.2.3 Broadcast方法实现
5.使用注意事项
5.1 调用Wait方法前未加锁
5.2 Wait方法接收到通知后，未重新检查条件变量
6.总结
1. 简介本文将介绍 Go 语言中的 sync.Cond 并发原语，包括 sync.Cond的基本使用方法、实现原理、使用注意事项以及常见的使用使用场景。能够更好地理解和应用 Cond 来实现 goroutine 之间的同步。
sync.Cond是Go语言标准库中的一个类型，代表条件变量。条件变量是用于多个goroutine之间进行同步和互斥的一种机制。sync.Cond可以用于等待和通知goroutine，以便它们可以在特定条件下等待或继续执行。
sync.Cond的定义如下，提供了Wait ,Singal,Broadcast以及NewCond方法
type Cond struct {
noCopy noCopy
// L is held while observing or changing the condition
L Locker
notify notifyList
checker copyChecker
}
func NewCond(l Locker) *Cond {}
func (c *Cond) Wait() {}
func (c *Cond) Signal() {}
func (c *Cond) Broadcast() {}
NewCond方法： 提供创建Cond实例的方法
Wait方法: 使当前线程进入阻塞状态，等待其他协程唤醒
Singal方法: 唤醒一个等待该条件变量的线程，如果没有线程在等待，则该方法会立即返回。
Broadcast方法: 唤醒所有等待该条件变量的线程，如果没有线程在等待，则该方法会立即返回。
当使用sync.Cond时，通常需要以下几个步骤：
定义一个互斥锁，用于保护共享数据；
创建一个sync.Cond对象，关联这个互斥锁；
在需要等待条件变量的地方，获取这个互斥锁，并使用Wait方法等待条件变量被通知；
在需要通知等待的协程时，使用Signal或Broadcast方法通知等待的协程。
最后，释放这个互斥锁。
2.4 使用例子下面是一个使用sync.Cond的简单示例，实现了一个生产者-消费者模型：
var (
// 1. 定义一个互斥锁
mu sync.Mutex
cond *sync.Cond
count int
)
func init() {
// 2.将互斥锁和sync.Cond进行关联
cond = sync.NewCond(&mu)
}
func worker(id int) {
// 消费者
for {
// 3. 在需要等待的地方,获取互斥锁，调用Wait方法等待被通知
mu.Lock()
// 这里会不断循环判断 是否有待消费的任务
for count == 0 {
cond.Wait() // 等待任务
}
count--
fmt.Printf("worker %d: 处理了一个任务\n", id)
// 5. 最后释放锁
mu.Unlock()
}
}
func main() {
// 启动5个消费者
for i := 1; i <= 5; i++ {
go worker(i)
}
for {
// 生产者
time.Sleep(1 * time.Second)
mu.Lock()
count++
// 4. 在需要等待的地方,获取互斥锁，调用BroadCast/Singal方法进行通知
cond.Broadcast()
mu.Unlock()
}
}
在这个示例中，创建一个生产者在生产任务，同时创建五个消费者来消费任务。当任务数为0时，此时消费者会调用Wait方法进入阻塞状态，等待生产者的通知。
当生产者产生任务后，使用Broadcast方法通知所有的消费者，唤醒处于阻塞状态的消费者，开始消费任务。这里使用sync.Cond实现多个协程之间的通信和同步。
这里的原因在于调用Wait方法前如果不加锁，有可能会出现竞态条件。
这里假设多个协程都处于等待状态，然后一个协程调用了Broadcast唤醒了其中一个或多个协程，此时这些协程都会被唤醒。
如下，假设调用Wait方法前没有加锁的话，那么所有协程都会去调用condition方法去判断是否满足条件，然后都通过验证，执行后续操作。
for !condition() {
c.Wait()
}
c.L.Lock()
// 满足条件情况下,执行的逻辑
c.L.Unlock()
此时会出现的情况为，本来是需要在满足condition方法的前提下，才能执行的操作。现在有可能的效果，为前面一部分协程执行时，还是满足condition条件的；但是后面的协程，尽管不满足condition条件，还是执行了后续操作，可能导致程序出错。
正确的用法应该是，在调用Wait方法前便加锁，那么即使多个协程被唤醒，一次也只会有一个协程判断是否满足condition条件，然后执行后续操作。这样子就不会出现多个协程同时判断，导致不满足条件，也执行后续操作的情况出现。
c.L.Lock()
for !condition() {
c.Wait()
}
// 满足条件情况下,执行的逻辑
c.L.Unlock()
3.使用场景
3.1 基本说明
sync.Cond是为了协调多个协程之间对共享数据的访问而设计的。使用sync.Cond的场景通常都涉及到对共享数据的操作，如果没有共享数据的操作，那么没有太大必要使用sync.Cond来进行协调。当然，如果存在重复唤醒的场景，即使没有对共享数据的操作，也是可以使用sync.Cond来进行协调的。
通常情况下，使用sync.Cond的场景为：多个协程需要访问同一份共享数据，需要等待某个条件满足后才能访问或修改这份共享数据。
在这些场景下，使用sync.Cond可以方便地实现对共享数据的协调，避免了多个协程之间的竞争和冲突，保证了共享数据的正确性和一致性。因此，如果没有涉及到共享数据的操作，就没有必要使用sync.Cond来进行协调。
下面举一个使用 sync.Cond 的例子，用它来实现生产者-消费者模型。生产者往items放置元素，当items满了之后，便进入等待状态，等待消费者唤醒。消费者从items中取数据，当items空了之后，便进入等待状态，等待生产者唤醒。
这里多个协程对同一份数据进行操作，且需要基于该数据判断是否唤醒其他协程或进入阻塞状态，来实现多个协程的同步和协调。sync.Cond就适合在这种场景下使用，其正是为这种场景设计的。
package main
import (
"fmt"
"sync"
"time"
)
type Queue struct {
items []int
cap int
lock sync.Mutex
cond *sync.Cond
}
func NewQueue(cap int) *Queue {
q := &Queue{
items: make([]int, 0),
cap: cap,
}
q.cond = sync.NewCond(&q.lock)
return q
}
func (q *Queue) Put(item int) {
q.lock.Lock()
defer q.lock.Unlock()
for len(q.items) == q.cap {
q.cond.Wait()
}
q.items = append(q.items, item)
q.cond.Broadcast()
}
func (q *Queue) Get() int {
q.lock.Lock()
defer q.lock.Unlock()
for len(q.items) == 0 {
q.cond.Wait()
}
item := q.items[0]
q.items = q.items[1:]
q.cond.Broadcast()
return item
}
func main() {
q := NewQueue(10)
// Producer
go func() {
for {
q.Put(i)
fmt.Printf("Producer: Put %d\n", i)
time.Sleep(100 * time.Millisecond)
}
}()
// Consumer
go func() {
for {
item := q.Get()
fmt.Printf("Consumer: Get %d\n", item)
time.Sleep(200 * time.Millisecond)
}
}()
wg.Wait()
}
3.2.2 需要重复唤醒的场景中使用
在某些场景中，由于不满足某种条件，此时协程进入阻塞状态，等待条件满足后，由其他协程唤醒，再继续执行。在整个流程中，可能会多次进入阻塞状态，多次被唤醒的情况。
比如上面生产者和消费者模型的例子，生产者可能会产生一批任务，然后唤醒消费者，消费者消费完之后，会进入阻塞状态，等待下一批任务的到来。所以这个流程中，协程可能多次进入阻塞状态，然后再多次被唤醒。
sync.Cond能够实现即使协程多次进入阻塞状态，也能重复唤醒该协程。所以，当出现需要实现重复唤醒的场景时，使用sync.Cond也是非常合适的。
在Sync.Cond存在一个通知队列，保存了所有处于等待状态的协程。通知队列定义如下:
type notifyList struct {
wait uint32
notify uint32
lock uintptr // key field of the mutex
head unsafe.Pointer
tail unsafe.Pointer
}
当调用Wait方法时，此时Wait方法会释放所持有的锁，然后将自己放到notifyList等待队列中等待。此时会将当前协程加入到等待队列的尾部，然后进入阻塞状态。
当调用Signal 时，此时会唤醒等待队列中的第一个协程，其他继续等待。如果此时没有处于等待状态的协程，调用Signal不会有其他作用，直接返回。当调用BoradCast方法时，则会唤醒notfiyList中所有处于等待状态的协程。
sync.Cond的代码实现比较简单，协程的唤醒和阻塞已经由运行时包实现了，sync.Cond的实现直接调用了运行时包提供的API。
Wait方法首先调用runtime_notifyListAd方法，将自己加入到等待队列中，然后释放锁，等待其他协程的唤醒。
func (c *Cond) Wait() {
// 将自己放到等待队列中
t := runtime_notifyListAdd(&c.notify)
// 释放锁
c.L.Unlock()
// 等待唤醒
runtime_notifyListWait(&c.notify, t)
// 重新获取锁
c.L.Lock()
}
4.2.2 Singal方法实现
Singal方法调用runtime_notifyListNotifyOne唤醒等待队列中的一个协程。
func (c *Cond) Signal() {
// 唤醒等待队列中的一个协程
runtime_notifyListNotifyOne(&c.notify)
}
4.2.3 Broadcast方法实现
Broadcast方法调用runtime_notifyListNotifyAll唤醒所有处于等待状态的协程。
func (c *Cond) Broadcast() {
// 唤醒等待队列中所有的协程
runtime_notifyListNotifyAll(&c.notify)
}
5.使用注意事项
5.1 调用Wait方法前未加锁
在上面2.5已经说明了，调用Sync.Cond方法前需要加锁，否则有可能出现竞态条件。而且，现有的sync.Cond的实现，如果在调用Wait方法前未加锁，此时会直接panic，下面是一个简单例子的说明:
package main
import (
"fmt"
"sync"
"time"
)
var (
count int
cond *sync.Cond
lk sync.Mutex
)
func main() {
cond = sync.NewCond(&lk)
wg := sync.WaitGroup{}
wg.Add(2)
go func() {
defer wg.Done()
for {
time.Sleep(time.Second)
count++
cond.Broadcast()
}
}()
go func() {
defer wg.Done()
for {
time.Sleep(time.Millisecond * 500)
//cond.L.Lock()
for count%10 != 0 {
cond.Wait()
}
t.Logf("count = %d", count)
//cond.L.Unlock()
}
}()
wg.Wait()
}
上面代码中，协程一每隔1s，将count字段的值自增1，然后唤醒所有处于等待状态的协程。协程二执行的条件为count的值为10的倍数，此时满足执行条件，唤醒后将会继续往下执行。
但是这里在调用sync.Wait方法前，没有先获取锁，下面是其执行结果，会抛出 fatal error: sync: unlock of unlocked mutex 错误，结果如下:
count = 0
fatal error: sync: unlock of unlocked mutex
因此，在调用Wait方法前，需要先获取到与sync.Cond关联的锁，否则会直接抛出异常。
调用sync.Wait方法，协程进入阻塞状态后被唤醒，没有重新检查条件变量，此时有可能仍然处于不满足条件变量的场景下。然后直接执行后续操作，有可能会导致程序出错。下面举一个简单的例子:
package main
import (
"fmt"
"sync"
"time"
)
var (
count int
cond *sync.Cond
lk sync.Mutex
)
func main() {
cond = sync.NewCond(&lk)
wg := sync.WaitGroup{}
wg.Add(3)
go func() {
defer wg.Done()
for {
time.Sleep(time.Second)
cond.L.Lock()
// 将flag 设置为true
flag = true
// 唤醒所有处于等待状态的协程
cond.Broadcast()
cond.L.Unlock()
}
}()
for i := 0; i < 2; i++ {
go func(i int) {
defer wg.Done()
for {
time.Sleep(time.Millisecond * 500)
cond.L.Lock()
// 不满足条件，此时进入等待状态
if !flag {
cond.Wait()
}
// 被唤醒后，此时可能仍然不满足条件
fmt.Printf("协程 %d flag = %t", i, flag)
flag = false
cond.L.Unlock()
}
}(i)
}
wg.Wait()
}
在这个例子，我们启动了一个协程，定时将flag设置为true，相当于每隔一段时间，便满足执行条件，然后唤醒所有处于等待状态的协程。
然后又启动了两个协程，在满足条件的前提下，开始执行后续操作，但是这里协程被唤醒后，没有重新检查条件变量，具体看第39行。这里会出现的场景是，第一个协程被唤醒后，此时执行后续操作，然后将flag重新设置为false，此时已经不满足条件了。之后第二个协程唤醒后，获取到锁，没有重新检查此时是否满足执行条件，直接向下执行，这个就和我们预期不符，可能会导致程序出错，代码执行效果如下:
协程 1 flag = true
协程 0 flag = false
协程 1 flag = true
协程 0 flag = false
可以看到，此时协程0执行时，flag的值均为false,说明此时其实并不符合执行条件，可能会导致程序出错。因此正确用法应该像下面这样子，被唤醒后，需要重新检查条件变量，满足条件之后才能继续向下执行。
c.L.Lock()
// 唤醒后,重新检查条件变量是否满足条件
for !condition() {
c.Wait()
}
// 满足条件情况下,执行的逻辑
c.L.Unlock()
6.总结
本文介绍了 Go 语言中的 sync.Cond 并发原语，它是用于实现 goroutine 之间的同步的重要工具。我们首先学习了 sync.Cond 的基本使用方法，包括创建和使用条件变量、使用Wait和Signal/Broadcast方法等。
接着，我们对 sync.Cond 的使用场景进行了说明，如同步和协调多个协程之间共享资源等。
在接下来的部分中，我们介绍了 sync.Cond 的实现原理，主要是对等待队列的使用，从而sync.Cond有更好的理解，能够更好得使用它。同时，我们也讲述了使用sync.Cond的注意事项，如调用Wait方法前需要加锁等。
基于以上内容，本文完成了对 sync.Cond 的介绍，希望能够帮助大家更好地理解和使用Go语言中的并发原语。
更多关于Go语言sync.Cond原理的资料请关注软件开发网其它相关文章！
