详解Go条件变量cond的使用

在 Go 语言中，条件变量（sync.Cond）是一种用于实现线程间同步的工具。它允许一个或多个 goroutine 等待某个条件的发生。条件变量通常与互斥锁（sync.Mutex）结合使用，以确保对共享数据的安全访问。本文将详细介绍 Go 中条件变量的使用，包括实现、常见使用陷阱以及具体实例。

1. 条件变量的概念

条件变量的作用是让 goroutine 可以在某个条件不满足时进行等待，直到其他 goroutine 通知它们条件已经满足。常见的应用场景包括生产者-消费者模式、任务调度等。

1.1. 条件变量的结构

Go 中的条件变量通过 sync.Cond 类型提供。它包含以下几个重要的方法：

Wait(): 使调用者等待条件变量被唤醒。
Signal(): 唤醒一个等待的 goroutine。
Broadcast(): 唤醒所有等待的 goroutine。

1.2. 使用示例

下面是一个简单的条件变量示例：

package main

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

var (
    mu      sync.Mutex
    cond    = sync.NewCond(&mu)
    ready   = false
)

func worker() {
    mu.Lock()
    for !ready {
        cond.Wait() // 等待条件变量
    }
    fmt.Println("Worker: condition met, proceeding...")
    mu.Unlock()
}

func main() {
    go worker()

    time.Sleep(1 * time.Second) // 等待一段时间，确保 worker goroutine 已经运行
    mu.Lock()
    ready = true
    cond.Signal() // 通知 worker goroutine
    mu.Unlock()

    time.Sleep(1 * time.Second) // 等待一段时间以观察输出
}

在上述代码中，worker 函数首先锁定互斥锁，然后检查 ready 变量。如果 ready 为 false，则调用 cond.Wait() 进入等待状态。main 函数在等待一段时间后设置 ready 为 true，并调用 cond.Signal() 唤醒 worker 函数。

2. 条件变量的实现

条件变量的实现本质上是一个包含了互斥锁和一个等待队列的结构。sync.Cond 的实现大致如下：

type Cond struct {
    nof       int32             // 等待的 goroutine 数量
    waiters   sync.WaitGroup    // 用于等待的 goroutine
    mu        sync.Mutex        // 互斥锁
    notify    chan struct{}     // 通知通道
}

2.1. Wait 方法

Wait 方法的主要逻辑是释放互斥锁，并将当前 goroutine 添加到等待队列中。当条件被满足时，它会被唤醒。

2.2. Signal 方法

Signal 方法用于唤醒一个等待的 goroutine。如果没有 goroutine 在等待，则该方法不会有任何效果。

2.3. Broadcast 方法

Broadcast 方法用于唤醒所有在条件变量上等待的 goroutine。

3. 使用陷阱

使用条件变量时需要注意以下几点：

3.1. 锁的使用

在调用 Wait、Signal 和 Broadcast 方法时，必须在互斥锁保护下进行。这是确保数据一致性的重要步骤。

3.2. 误用条件变量

如果不小心在条件变量不满足时调用 Signal 或 Broadcast，可能会导致其他 goroutine 错误地被唤醒。

3.3. 资源泄漏

如果在 goroutine 中没有及时释放锁，可能导致资源泄漏或死锁。

4. 实际应用示例

下面是一个实际的生产者-消费者模型示例，展示了条件变量的使用。

package main

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

type BoundedBuffer struct {
    buf   []int
    size  int
    mu    sync.Mutex
    notEmpty *sync.Cond
    notFull  *sync.Cond
}

func NewBoundedBuffer(size int) *BoundedBuffer {
    bb := &BoundedBuffer{
        buf:  make([]int, 0, size),
        size: size,
    }
    bb.notEmpty = sync.NewCond(&bb.mu)
    bb.notFull = sync.NewCond(&bb.mu)
    return bb
}

func (bb *BoundedBuffer) Put(item int) {
    bb.mu.Lock()
    defer bb.mu.Unlock()
    for len(bb.buf) == bb.size {
        bb.notFull.Wait() // 等待缓冲区有空间
    }
    bb.buf = append(bb.buf, item)
    bb.notEmpty.Signal() // 通知消费者
}

func (bb *BoundedBuffer) Get() int {
    bb.mu.Lock()
    defer bb.mu.Unlock()
    for len(bb.buf) == 0 {
        bb.notEmpty.Wait() // 等待缓冲区有数据
    }
    item := bb.buf[0]
    bb.buf = bb.buf[1:]
    bb.notFull.Signal() // 通知生产者
    return item
}

func producer(bb *BoundedBuffer, id int) {
    for i := 0; i < 5; i++ {
        item := i + id*100 // 区分不同生产者生成的产品
        fmt.Printf("Producer %d: producing %d\n", id, item)
        bb.Put(item)
        time.Sleep(time.Millisecond * 500) // 模拟生产耗时
    }
}

func consumer(bb *BoundedBuffer, id int) {
    for i := 0; i < 5; i++ {
        item := bb.Get()
        fmt.Printf("Consumer %d: consuming %d\n", id, item)
        time.Sleep(time.Millisecond * 700) // 模拟消费耗时
    }
}

func main() {
    bb := NewBoundedBuffer(3)

    for i := 0; i < 2; i++ {
        go producer(bb, i)
    }

    for i := 0; i < 3; i++ {
        go consumer(bb, i)
    }

    time.Sleep(10 * time.Second) // 等待一段时间让所有 goroutine 完成
}

4.1. 代码解析

BoundedBuffer 结构: 定义了一个带有上限的缓冲区，包含一个切片用于存储数据，以及两个条件变量 notEmpty 和 notFull。
Put 方法: 生产者向缓冲区添加数据，若缓冲区满则等待 notFull。
Get 方法: 消费者从缓冲区获取数据，若缓冲区为空则等待 notEmpty。
生产者和消费者: 分别启动两个生产者和三个消费者 goroutine。