条件变量与生产者消费者模型 (Condition Variable & Producer-Consumer)

在多线程编程中，互斥锁解决了“抢占数据”的问题，但现实中很多场景并不是简单地争夺数据，而是涉及到线程之间的协作与等待： - 线程 A 负责从网络下载文件，线程 B 负责对文件进行解压分析。线程 B 必须等待线程 A 下载完毕才能开工； - 线程池里的工作线程，在没有任务时应该安静地“休眠等待”，而不是疯狂空转；一旦有新任务提交，应该立即被“拍醒开工”。

很多初学者可能会写出下面的轮询代码：

// ❌ 极度恶劣的忙等待 (Busy Waiting)！
while (!ready) {
    // 空循环，CPU 单核使用率瞬间飙满 100%！
}

这种做法不仅极其浪费 CPU 和电池电量，还会挤占其它正常工作线程的时间片。解决线程协作与等待的真正正规军，就是 条件变量（Condition Variable）。

---

1. 条件变量的核心运作机制

条件变量是操作系统与 C++ 标准库提供的一种线程同步原语。它的核心逻辑是：

“条件不满足？那我先把锁交出来，自己进入梦乡挂起；等你们把数据准备好了，再主动把我唤醒！”

sequenceDiagram
    autonumber
    actor Consumer as 消费者线程 (等待数据)
    participant Lock as 互斥锁 (unique_lock)
    participant CV as 条件变量 (std::condition_variable)
    actor Producer as 生产者线程 (生产数据)

    Consumer->>Lock: 1. 获取互斥锁
    Consumer->>Consumer: 2. 检查条件: 队列为空！
    Consumer->>CV: 3. 调用 cv.wait(lock)<br>【关键】原子地释放互斥锁，并进入睡眠挂起！
    Note over Consumer: 消费者沉睡中，不消耗任何 CPU 算力

    Producer->>Lock: 4. 趁着锁被释放，生产者成功抢到锁
    Producer->>Producer: 5. 生产数据并放入队列
    Producer->>Lock: 6. 释放互斥锁
    Producer->>CV: 7. 发送唤醒信号: cv.notify_one()

    CV-->>Consumer: 8. 唤醒消费者线程！
    Note over Consumer: 消费者苏醒，重新自动争抢并获取互斥锁
    Consumer->>Consumer: 9. 消费数据，离开临界区

为什么 cv.wait() 必须绑定 std::unique_lock？

初学者常常疑惑：为什么 cv.wait() 拒绝接受最轻量的 std::lock_guard，强制要求传入 std::unique_lock？ - 因为在挂起当前线程的一瞬间，wait() 必须在底层自动调用 lock.unlock() 释放互斥锁（否则别的线程永远拿不到锁，也就永远无法生产数据来满足条件，造成死锁！）； - 当线程被唤醒时，wait() 又必须在底层自动调用 lock.lock() 重新夺回互斥锁，保护后续对临界区数据的安全读取； - std::lock_guard 无法在中途手动解锁和重新加锁，唯有 std::unique_lock 具备这种灵活性。

---

2. 初学者必踩的两大深坑：虚假唤醒与信号丢失

条件变量是并发编程中最容易写出偶发性 Bug 的重灾区，其中最著名的就是虚假唤醒与信号丢失。

2.1 幽灵唤醒：虚假唤醒 (Spurious Wakeup)

你可能会理所当然地认为：“既然我的线程从 cv.wait() 醒来了，说明肯定有其他线程调用了 notify 且条件已经成熟了！” 错！在操作系统底层，线程完全可能在没有收到任何通知的情况下，由于硬件中断或调度器异常自发醒来，这就是虚假唤醒（Spurious Wakeup）。

// ❌ 极度危险的 if 检查！
std::unique_lock<std::mutex> lock(mtx);
if (data_queue.empty()) {
    cv.wait(lock); // 如果遭遇虚假唤醒，直接往下走！
}
auto item = data_queue.front(); // 灾难！队列依然为空，访问空指针崩溃！

破除虚假唤醒的黄金铁律：

永远用 while 循环 或者 带谓词（Predicate）的重载版本 重新检验条件！

// 正确写法 A: while 循环重验
while (data_queue.empty()) {
    cv.wait(lock);
}

// 正确写法 B: 现代 C++ 推荐的谓词形式 (语义完全等价于上述 while 循环)
cv.wait(lock, [] { return !data_queue.empty(); });

---

2.2 信号遗失：信号丢失 (Lost Wakeup)

如果在消费者进入 wait 挂起之前，生产者就已经发出了 cv.notify_one()，这个通知信号就会如同把石子扔进空无一人的深井——被系统直接静默丢弃！ 随后消费者慢吞吞地进入 cv.wait()，由于再也没有生产者发送通知，消费者将陷入无休止的永久沉睡。

防御铁律：

1. 条件状态的检查与修改，必须严格处于互斥锁的保护之下；
2. 确保在调用 notify 之前，状态标识已经完成变更。

---

3. 经典工业级实战：线程安全阻塞有界队列

生产者-消费者模型（Producer-Consumer Pattern） 是全行业使用最广泛的并发解耦架构。 以下是一个基于双条件变量实现的工业级线程安全有界阻塞队列（Bounded Blocking Queue）完整源码：

#include <iostream>
#include <queue>
#include <mutex>
#include <condition_variable>
#include <thread>
#include <vector>

template <typename T>
class BoundedBlockingQueue {
private:
    std::queue<T> queue_;
    const size_t capacity_;       // 队列容量上限
    mutable std::mutex mtx_;      // 保护队列的互斥锁
    std::condition_variable not_full_;  // 队列未满条件变量 (供生产者等待)
    std::condition_variable not_empty_; // 队列非空条件变量 (供消费者等待)
    bool closed_ = false;         // 队列关闭优雅停机标志

public:
    explicit BoundedBlockingQueue(size_t capacity) : capacity_(capacity) {}

    // 生产者入队操作
    bool push(T item) {
        std::unique_lock<std::mutex> lock(mtx_);
        // 如果队列已满且未关闭，生产者挂起等待
        not_full_.wait(lock, [this] {
            return queue_.size() < capacity_ || closed_;
        });

        if (closed_) return false; // 队列已关闭，拒绝写入

        queue_.push(std::move(item));
        // 唤醒可能正在沉睡的消费者
        not_empty_.notify_one();
        return true;
    }

    // 消费者出队操作
    bool pop(T& value) {
        std::unique_lock<std::mutex> lock(mtx_);
        // 如果队列为空且未关闭，消费者挂起等待
        not_empty_.wait(lock, [this] {
            return !queue_.empty() || closed_;
        });

        if (queue_.empty() && closed_) {
            return false; // 队列已清空且已关闭
        }

        value = std::move(queue_.front());
        queue_.pop();
        // 唤醒可能正在等待空间的生产者
        not_full_.notify_one();
        return true;
    }

    // 优雅关闭队列
    void close() {
        {
            std::lock_guard<std::mutex> lock(mtx_);
            closed_ = true;
        }
        // 广播唤醒所有正在等待的生产者和消费者，让它们安全退出
        not_full_.notify_all();
        not_empty_.notify_all();
    }
};

// 测试示例
int main() {
    BoundedBlockingQueue<int> bb_queue(5); // 容量上限为 5

    // 启动 2 个生产者线程
    std::vector<std::thread> producers;
    for (int i = 0; i < 2; ++i) {
        producers.emplace_back([&bb_queue, i]() {
            for (int j = 0; j < 10; ++j) {
                int item = i * 100 + j;
                if (!bb_queue.push(item)) break;
                std::cout << "[生产者 " << i << "] 写入数据: " << item << "\n";
            }
        });
    }

    // 启动 3 个消费者线程
    std::vector<std::thread> consumers;
    for (int i = 0; i < 3; ++i) {
        consumers.emplace_back([&bb_queue, i]() {
            int val;
            while (bb_queue.pop(val)) {
                std::cout << "  -> [消费者 " << i << "] 成功取出: " << val << "\n";
            }
        });
    }

    for (auto& p : producers) p.join();
    bb_queue.close(); // 生产完毕，优雅关闭
    for (auto& c : consumers) c.join();

    std::cout << "所有生产与消费任务圆满完成！\n";
    return 0;
}

---

4. C++20 现代同步原语扩展

随着 C++20 的发布，标准库在 <semaphore>、<latch>、<barrier> 中补充了更轻量、更高维度的线程同步工具，进一步简化了传统的复杂条件变量设计：

flowchart TD
    C20["C++20 现代同步原语"]

    C20 --> Sem["std::counting_semaphore<br>计数信号量: 替代经典的 PV 操作，轻松控制资源并发访问配额"]
    C20 --> Latch["std::latch (倒计时门栓)<br>一次性同步计数器: 等待 N 个子线程全部完成初始化后，大门打开，全员齐步走"]
    C20 --> Barrier["std::barrier (循环屏障)<br>多阶段循环同步: 多个线程分阶段并行计算，每一轮所有线程到齐后自动重置，进入下一轮"]

• std::latch（门栓）：非常适合“主线程等待 N 个工作线程完成数据预热，再统一对外提供服务”的场景。内部维护一个初始计数，子线程调用 count_down()，主线程调用 wait()，一次性使用不可复位；
• std::barrier（循环屏障）：多用于高性能科学计算（如并行矩阵乘法、物理模拟迭代）。线程到达屏障点后挂起，直到参与的所有线程均到达，系统执行一个可选的阶段完成回调，接着自动复位屏障，所有线程进入下一轮循环！