1. 简介
本文将介绍 Go 语言中的 sync.Cond
并发原语,包括 sync.Cond
的基本使用方法、实现原理、使用注意事项以及常见的使用使用场景。能够更好地理解和应用 Cond 来实现 goroutine 之间的同步。
2. 基本使用
2.1 定义
sync.Cond
是Go语言标准库中的一个类型,代表条件变量。条件变量是用于多个goroutine之间进行同步和互斥的一种机制。sync.Cond
可以用于等待和通知goroutine,以便它们可以在特定条件下等待或继续执行。
2.2 方法说明
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
方法: 唤醒所有等待该条件变量的线程,如果没有线程在等待,则该方法会立即返回。
2.3 使用方式
当使用sync.Cond
时,通常需要以下几个步骤:
- 定义一个互斥锁,用于保护共享数据;
- 创建一个
sync.Cond
对象,关联这个互斥锁; - 在需要等待条件变量的地方,获取这个互斥锁,并使用
Wait
方法等待条件变量被通知; - 在需要通知等待的协程时,使用
Signal
或Broadcast
方法通知等待的协程。 - 最后,释放这个互斥锁。
下面是一个简单的代码的示例,展示了大概的代码结构:
var (
// 1. 定义一个互斥锁
mu sync.Mutex
cond *sync.Cond
count int
)
func init() {
// 2.将互斥锁和sync.Cond进行关联
cond = sync.NewCond(&mu)
}
go func(){
// 3. 在需要等待的地方,获取互斥锁,调用Wait方法等待被通知
mu.Lock()
// 这里会不断循环判断 是否满足条件
for !condition() {
cond.Wait() // 等待任务
}
mu.Unlock()
}
go func(){
// 执行业务逻辑
// 4. 满足条件,此时调用Broadcast唤醒处于等待状态的协程
cond.Broadcast()
}
2.4 使用例子
下面通过描述net/http
中的 connReader
,来展示使用sync.Cond
实现阻塞等待通知的机制。这里我们不需要理解太多,只需要知道connReader
下面两个方法:
func (cr *connReader) Read(p []byte) (n int, err error) {}
func (cr *connReader) abortPendingRead() {}
Read
方法则是用于从HTTP
连接中读取数据,不允许并发访问的。而abortPendingRead
则是用于终止正在读取的连接。
从abortPendingRead
方法的语意来看,是需要成功终止其他协程进行数据的读取之后,才能正常返回,也就是此时没有协程再继续读取数据了,才可以返回。
那abortPendingRead
如何得知是否还有协程在读取数据呢,其实是可以通过定时轮训connReader
的状态,从而判断当前Read方法是否仍在读取数据。但是定时轮训效率太低,可能会造成cpu的大量空转。更好的方式,应该是让协程进入阻塞状态,然后等条件满足了,其他协程再来唤醒当前协程,然后再继续运行下去。
这个其实就是sync.Cond
设计的用途,当不满足运行条件时,先进入阻塞状态,等待条件满足时,再由其他协程来唤醒,然后再继续运行下去,能够提高程序的执行效率。其中Wait
方法便是让协程进入阻塞状态,而Singal
和Boardcast
便是唤醒处于阻塞状态的协程,告知其条件满足了,可以继续向下执行了。
回到我们connReader
的例子,我们使用sync.Cond
实现阻塞等待通知的效果。
type connReader struct {
// 是否正在读取数据
inRead bool
mu sync.Mutex // guards following
cond *sync.Cond
}
func (cr *connReader) abortPendingRead() {
if !cr.inRead{
return
}
//1. 通过一定手段,让Read方法中断
cr.mu.Lock()
// 判断Read方法是否仍然在读取数据
for cr.inRead {
//2. 此时Read方法仍然在读取数据, 不满足条件,等待通知
cr.cond.Wait()
}
cr.mu.Unlock()
}
func (cr *connReader) Read(p []byte) (n int, err error) {
cr.mu.Lock()
cr.inRead = true
// 1. 读取数据
// 2. abortPendingRead通过某种手段,让Read方法中断
cr.inRead = false
cr.mu.Unlock()
// 3. 现在已经满足abortPendingRead继续执行下去的条件了,可以唤醒abortPendingRead协程了
cond.Boardcast()
}
这里abortPendingRead
方法首先判断是否还在读取数据,是的话,调用Wait
方法进入阻塞状态,等待条件满足后继续执行。
对于Read
方法,因为其不运行并发访问,当其将退出时,说明此时已经没有协程在读取数据了,满足abortPendingRead
继续执行下去的条件了,此时可以调用Boardcast
来唤醒等待条件满足的协程。之后调用abortPendingRead
方法的协程此时能够接收到通知,便能够顺利被唤醒,从而正确返回。
这里便展示了一个简单的,使用sync.Cond
实现阻塞等待通知的例子。
3. 原理
3.1 基本原理
在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
中所有处于等待