Cond条件变量 - 元素码农

发布时间: 2025-04-24 23:09

↑

# Cond条件变量

## 简介

sync.Cond是Go语言标准库中提供的条件变量实现，它是一种同步原语，用于在满足特定条件时唤醒等待的goroutine。条件变量总是与互斥锁一起使用，用于协调多个goroutine之间的协作。

条件变量的主要用途是让一组goroutine在某个条件成立时被唤醒，而不是通过轮询或定时检查来等待条件满足，这样可以避免CPU资源的浪费，提高程序的效率。

## 基本用法

```go
package main

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

func main() {
    // 创建互斥锁
    var mu sync.Mutex
    // 基于互斥锁创建条件变量
    cond := sync.NewCond(&mu)
    
    // 共享数据
    queue := make([]int, 0, 10)
    
    // 消费者goroutine
    go func() {
        for {
            mu.Lock()
            // 当队列为空时等待
            for len(queue) == 0 {
                fmt.Println("消费者: 队列为空，等待生产者...")
                cond.Wait() // 等待条件变量通知
                fmt.Println("消费者: 被唤醒")
            }
            
            // 消费数据
            item := queue[0]
            queue = queue[1:]
            fmt.Printf("消费者: 消费了 %d，队列长度 %d\n", item, len(queue))
            
            mu.Unlock()
            time.Sleep(time.Second) // 模拟处理时间
        }
    }()
    
    // 生产者
    for i := 1; i <= 5; i++ {
        time.Sleep(2 * time.Second) // 模拟生产间隔
        
        mu.Lock()
        // 生产数据
        queue = append(queue, i)
        fmt.Printf("生产者: 生产了 %d，队列长度 %d\n", i, len(queue))
        // 通知等待的消费者
        cond.Signal()
        mu.Unlock()
    }
    
    // 等待足够时间让消费者处理完
    time.Sleep(10 * time.Second)
}
```

## 内部结构

sync.Cond的内部结构相对简单：

```go
type Cond struct {
    noCopy noCopy
    L Locker
    notify  notifyList
    checker copyChecker
}
```

- `noCopy`：一个空结构体，用于防止Cond被复制使用
- `L`：互斥锁接口，用于保护条件和共享数据
- `notify`：通知列表，用于管理等待的goroutine
- `checker`：用于检测Cond是否被复制

其中，notifyList是条件变量的核心部分，它维护了一个等待队列，用于存储等待条件的goroutine。

## 主要方法

sync.Cond提供了三个主要方法：

### NewCond

```go
func NewCond(l Locker) *Cond
```

NewCond创建一个新的条件变量，需要传入一个实现了Locker接口的互斥锁（通常是sync.Mutex或sync.RWMutex）。

### Wait

```go
func (c *Cond) Wait()
```

Wait方法会原子地解锁c.L并阻塞当前goroutine，将其加入到等待队列中。当被唤醒时，Wait会重新锁定c.L并返回。

**注意**：调用Wait方法前必须已经锁定c.L，Wait返回后c.L仍然是锁定状态。

### Signal

```go
func (c *Cond) Signal()
```

Signal方法唤醒等待队列中的一个goroutine。如果没有goroutine在等待，则什么也不做。

### Broadcast

```go
func (c *Cond) Broadcast()
```

Broadcast方法唤醒等待队列中的所有goroutine。如果没有goroutine在等待，则什么也不做。

## 实现原理

### Wait方法实现

Wait方法的实现涉及到几个关键步骤：

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

1. 首先检查Cond是否被复制
2. 将当前goroutine添加到通知列表中，并获取一个令牌t
3. 解锁互斥锁，允许其他goroutine访问共享数据
4. 阻塞当前goroutine，等待被唤醒
5. 当被唤醒后，重新锁定互斥锁

这个过程确保了等待和唤醒的原子性，避免了丢失唤醒信号的问题。

### Signal方法实现

Signal方法的实现相对简单：

```go
func (c *Cond) Signal() {
    c.checker.check()
    runtime_notifyListNotifyOne(&c.notify)
}
```

它只是简单地唤醒通知列表中的一个等待者。

### Broadcast方法实现

Broadcast方法类似于Signal，但它会唤醒所有等待者：

```go
func (c *Cond) Broadcast() {
    c.checker.check()
    runtime_notifyListNotifyAll(&c.notify)
}
```

## 使用模式

### 生产者-消费者模式

条件变量最常见的用途是实现生产者-消费者模式：

```go
type Queue struct {
    cond *sync.Cond
    data []interface{}
    capacity int
}

func NewQueue(capacity int) *Queue {
    var mu sync.Mutex
    return &Queue{
        cond: sync.NewCond(&mu),
        data: make([]interface{}, 0, capacity),
        capacity: capacity,
    }
}

func (q *Queue) Put(item interface{}) {
    q.cond.L.Lock()
    defer q.cond.L.Unlock()
    
    // 当队列满时等待
    for len(q.data) == q.capacity {
        q.cond.Wait()
    }
    
    // 添加数据并通知等待的消费者
    q.data = append(q.data, item)
    q.cond.Signal()
}

func (q *Queue) Get() interface{} {
    q.cond.L.Lock()
    defer q.cond.L.Unlock()
    
    // 当队列空时等待
    for len(q.data) == 0 {
        q.cond.Wait()
    }
    
    // 获取数据并通知等待的生产者
    item := q.data[0]
    q.data = q.data[1:]
    q.cond.Signal()
    
    return item
}
```

### 多生产者-多消费者

当有多个生产者和消费者时，通常使用Broadcast而不是Signal：

```go
func (q *Queue) Put(item interface{}) {
    q.cond.L.Lock()
    defer q.cond.L.Unlock()
    
    // 当队列满时等待
    for len(q.data) == q.capacity {
        q.cond.Wait()
    }
    
    // 添加数据并通知所有等待的消费者
    q.data = append(q.data, item)
    q.cond.Broadcast()
}

func (q *Queue) Get() interface{} {
    q.cond.L.Lock()
    defer q.cond.L.Unlock()
    
    // 当队列空时等待
    for len(q.data) == 0 {
        q.cond.Wait()
    }
    
    // 获取数据并通知所有等待的生产者
    item := q.data[0]
    q.data = q.data[1:]
    q.cond.Broadcast()
    
    return item
}
```

### 使用for循环而非if语句

在使用条件变量时，应该始终使用for循环而不是if语句来检查条件：

```go
// 正确的做法
for !condition() {
    cond.Wait()
}

// 错误的做法
if !condition() {
    cond.Wait()
}
```

这是因为Wait返回并不意味着条件一定满足了，可能是由于：

1. 虚假唤醒（spurious wakeup）：goroutine可能在没有收到Signal或Broadcast的情况下被唤醒
2. 条件竞争：在一个goroutine被唤醒但还没有获得锁的时间窗口内，另一个goroutine可能改变了条件

使用for循环可以确保在条件不满足时继续等待。

## 使用注意事项

1. **锁的状态**：调用Wait前必须持有锁，Wait返回后锁仍然是持有状态

2. **避免死锁**：确保在适当的时候调用Signal或Broadcast，否则等待的goroutine可能永远不会被唤醒

3. **条件检查**：使用for循环而不是if语句来检查条件

4. **Signal vs Broadcast**：
   - 当只有一个goroutine需要被唤醒时，使用Signal
   - 当多个goroutine可能都需要被唤醒时，使用Broadcast
   - 如果不确定，使用Broadcast更安全，但可能导致不必要的唤醒

5. **不要复制Cond**：sync.Cond包含指向运行时的指针，复制可能导致不可预期的行为

## 实际应用场景

### 实现工作池

```go
type WorkPool struct {
    cond      *sync.Cond
    tasks     []func()
    workers   int
    quit      bool
}

func NewWorkPool(workers int) *WorkPool {
    var mu sync.Mutex
    return &WorkPool{
        cond:    sync.NewCond(&mu),
        tasks:   make([]func(), 0),
        workers: workers,
        quit:    false,
    }
}

func (p *WorkPool) Start() {
    for i := 0; i < p.workers; i++ {
        go func(id int) {
            for {
                p.cond.L.Lock()
                for len(p.tasks) == 0 && !p.quit {
                    p.cond.Wait()
                }
                
                if p.quit {
                    p.cond.L.Unlock()
                    return
                }
                
                // 获取任务
                task := p.tasks[0]
                p.tasks = p.tasks[1:]
                p.cond.L.Unlock()
                
                // 执行任务
                task()
            }
        }(i)
    }
}

func (p *WorkPool) AddTask(task func()) {
    p.cond.L.Lock()
    defer p.cond.L.Unlock()
    
    p.tasks = append(p.tasks, task)
    p.cond.Signal() // 唤醒一个工作者
}

func (p *WorkPool) Stop() {
    p.cond.L.Lock()
    defer p.cond.L.Unlock()
    
    p.quit = true
    p.cond.Broadcast() // 唤醒所有工作者
}
```

### 实现并发控制

```go
type Barrier struct {
    cond      *sync.Cond
    count     int
    threshold int
}

func NewBarrier(threshold int) *Barrier {
    var mu sync.Mutex
    return &Barrier{
        cond:      sync.NewCond(&mu),
        threshold: threshold,
    }
}

func (b *Barrier) Wait() {
    b.cond.L.Lock()
    defer b.cond.L.Unlock()
    
    b.count++
    if b.count == b.threshold {
        // 达到阈值，重置计数并唤醒所有等待者
        b.count = 0
        b.cond.Broadcast()
    } else {
        // 等待其他goroutine到达
        b.cond.Wait()
    }
}
```

### 实现读写锁

条件变量可以用来实现自定义的读写锁：

```go
type RWLock struct {
    cond      *sync.Cond
    readers   int
    writing   bool
}

func NewRWLock() *RWLock {
    var mu sync.Mutex
    return &RWLock{
        cond: sync.NewCond(&mu),
    }
}

func (rw *RWLock) RLock() {
    rw.cond.L.Lock()
    defer rw.cond.L.Unlock()
    
    // 当有写锁时等待
    for rw.writing {
        rw.cond.Wait()
    }
    
    rw.readers++
}

func (rw *RWLock) RUnlock() {
    rw.cond.L.Lock()
    defer rw.cond.L.Unlock()
    
    rw.readers--
    if rw.readers == 0 {
        // 唤醒等待的写操作
        rw.cond.Signal()
    }
}

func (rw *RWLock) Lock() {
    rw.cond.L.Lock()
    defer rw.cond.L.Unlock()
    
    // 当有读锁或写锁时等待
    for rw.writing || rw.readers > 0 {
        rw.cond.Wait()
    }
    
    rw.writing = true
}

func (rw *RWLock) Unlock() {
    rw.cond.L.Lock()
    defer rw.cond.L.Unlock()
    
    rw.writing = false
    // 唤醒所有等待的读写操作
    rw.cond.Broadcast()
}
```

## 与其他同步原语的比较

| 同步原语 | 主要用途 | 适用场景 |
|---------|---------|----------|
| Mutex | 互斥访问共享资源 | 简单的互斥需求 |
| RWMutex | 读写分离 | 读多写少的场景 |
| Cond | 条件等待和通知 | 需要在特定条件下唤醒goroutine |
| WaitGroup | 等待一组goroutine完成 | 并行任务的同步点 |
| Channel | 通信和同步 | CSP模式，数据传递 |

## 性能考虑

条件变量在某些场景下比轮询或定时检查更高效：

1. **避免忙等待**：条件变量让goroutine在条件不满足时进入休眠状态，避免了CPU资源的浪费

2. **精确唤醒**：只有在条件可能满足时才唤醒等待的goroutine，减少了不必要的上下文切换

3. **批量唤醒**：Broadcast可以一次唤醒所有等待的goroutine，适合于条件变化影响多个等待者的场景

然而，条件变量也有一些性能上的考虑：

1. **锁的开销**：条件变量总是与互斥锁一起使用，频繁的锁操作可能成为性能瓶颈

2. **唤醒延迟**：从被唤醒到实际获得锁并执行可能有一定延迟

3. **不必要的唤醒**：Broadcast会唤醒所有等待的goroutine，即使只有一个能够继续执行，这可能导致不必要的上下文切换

## 总结

sync.Cond是Go语言中一个强大而灵活的同步原语，它通过条件等待和通知机制，为复杂的并发协作提供了支持。与互斥锁、通道等其他同步机制相比，条件变量在某些场景下提供了更自然、更高效的解决方案。

理解条件变量的工作原理和使用模式，有助于我们在适当的场景选择它，并正确地使用它来解决并发问题。特别是在生产者-消费者、资源池管理、并发控制等场景中，条件变量往往能够提供简洁而高效的实现。