条件变量与生产者-消费者模式

使用条件变量高效地协调线程。构建线程安全的生产者-消费者队列，处理虚假唤醒，并避免丢失唤醒问题。

使用条件变量进行线程协调

条件变量让线程可以等待某个条件变为真，而无需忙等循环。线程不再使用 while (!ready) { /* 消耗 CPU */ } 这样的自旋方式，而是进入休眠状态，由另一个线程调用 notify_one() 或 notify_all() 来唤醒它。

std::condition_variable 只与 std::unique_lock 配合使用。模式始终如下：

1. 锁定互斥锁。
2. 检查条件 ── 如果为假，调用 wait()，它会原子地释放互斥锁并挂起线程。
3. 当被通知（且条件为真）时，互斥锁被重新获取，执行继续。

通知线程在同一个互斥锁下修改共享状态，然后调用 notify_one() 或 notify_all()。

条件变量的基本用法

以下是基本模式：一个线程等待数据，另一个线程提供数据。

condition_variable_basic.cpp

#include <iostream>
#include <thread>
#include <mutex>
#include <condition_variable>
#include <string>

std::mutex mtx;
std::condition_variable cv;
std::string data;
bool ready = false;

void consumer() {
    std::unique_lock<std::mutex> lock(mtx);
    // Wait until ready is true (handles spurious wakeups)
    cv.wait(lock, []{ return ready; });
    std::cout << "Consumer received: " << data << "\n";
}

void producer() {
    {
        std::lock_guard<std::mutex> lock(mtx);
        data = "Hello from producer";
        ready = true;
    }  // unlock before notify to avoid waking thread into contention
    cv.notify_one();
}

int main() {
    std::thread t1(consumer);
    std::thread t2(producer);
    t1.join();
    t2.join();
    return 0;
}

常见陷阱

条件变量可能会在没有发送任何通知的情况下唤醒线程 ── 这被称为虚假唤醒。C++ 标准允许这种行为（在实际系统中很常见），这是出于性能考虑。你必须始终使用带谓词形式的 wait() 或将其包装在 while 循环中：

// 正确 ── 谓词形式处理虚假唤醒
cv.wait(lock, []{ return ready; });

// 等价的手动循环
while (!ready) {
    cv.wait(lock);
}

// 错误 ── 随机唤醒，在没有数据的情况下继续执行
cv.wait(lock);  // 没有条件检查！

永远不要在没有谓词或外围 while 循环的情况下调用 cv.wait(lock)。

线程安全的有界缓冲区（生产者-消费者）

生产者-消费者模式是最常见的并发模式之一。生产者向共享缓冲区添加项目；消费者从中移除项目。有界缓冲区在满时阻塞生产者，在空时阻塞消费者。

bounded_queue.cpp

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

template <typename T>
class BoundedQueue {
    std::queue<T> queue_;
    std::mutex mtx_;
    std::condition_variable not_full_;
    std::condition_variable not_empty_;
    size_t capacity_;
public:
    explicit BoundedQueue(size_t cap) : capacity_(cap) {}

    void push(T item) {
        std::unique_lock<std::mutex> lock(mtx_);
        not_full_.wait(lock, [this]{ return queue_.size() < capacity_; });
        queue_.push(std::move(item));
        not_empty_.notify_one();
    }

    T pop() {
        std::unique_lock<std::mutex> lock(mtx_);
        not_empty_.wait(lock, [this]{ return !queue_.empty(); });
        T item = std::move(queue_.front());
        queue_.pop();
        not_full_.notify_one();
        return item;
    }
};

int main() {
    BoundedQueue<int> bq(5);  // capacity 5

    // Producer
    std::thread producer([&bq]() {
        for (int i = 0; i < 20; ++i) {
            bq.push(i);
            std::cout << "Produced: " << i << "\n";
        }
    });

    // Consumer
    std::thread consumer([&bq]() {
        for (int i = 0; i < 20; ++i) {
            int val = bq.pop();
            std::cout << "Consumed: " << val << "\n";
        }
    });

    producer.join();
    consumer.join();
    return 0;
}

丢失唤醒问题

丢失唤醒发生在 notify_one() 在消费者调用 wait() 之前被调用的情况。由于没有线程在等待，通知就丢失了 ── 消费者随后将永远休眠。

标准模式通过在休眠之前检查谓词来防止这种情况。如果生产者已经将条件设为真，谓词返回真，wait() 立即返回而不休眠。这就是谓词形式 cv.wait(lock, pred) 必不可少的原因 ── 它在每次唤醒前后都检查条件：

1. 锁定互斥锁。
2. 检查谓词 ── 如果已经为真，跳过等待。
3. 否则，原子地解锁并休眠。
4. 唤醒时，重新加锁并重新检查谓词。

只要生产者和消费者都用同一个互斥锁保护共享标志，就不会丢失唤醒。

notify_one 与 notify_all

notify_one() 唤醒一个等待线程；notify_all() 唤醒所有等待线程。当多个线程可能需要重新检查其条件时（例如关闭信号），使用 notify_all()；当恰好一个线程应该继续执行时（例如单个消费者获取任务），使用 notify_one()。

notify_all_shutdown.cpp

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

std::mutex mtx;
std::condition_variable cv;
bool shutdown = false;

void worker(int id) {
    std::unique_lock<std::mutex> lock(mtx);
    cv.wait(lock, []{ return shutdown; });
    std::cout << "Worker " << id << " shutting down\n";
}

int main() {
    std::vector<std::thread> workers;
    for (int i = 0; i < 5; ++i) {
        workers.emplace_back(worker, i);
    }

    std::this_thread::sleep_for(std::chrono::seconds(1));

    {
        std::lock_guard<std::mutex> lock(mtx);
        shutdown = true;
    }
    cv.notify_all();  // wake ALL workers for shutdown

    for (auto& w : workers) w.join();
    return 0;
}

最佳实践

始终使用 wait() 的谓词重载来防范虚假唤醒和丢失唤醒。在可能的情况下，在调用 notify_one()/notify_all() 之前解锁互斥锁 ── 这样可以避免被唤醒的线程立即在互斥锁上阻塞。当只需要唤醒一个线程时优先使用 notify_one()，对于广播场景（如关闭）使用 notify_all()。永远不要用 lock_guard 等待条件变量 ── 你需要 unique_lock。

关键要点

std::condition_variable 让线程休眠直到条件为真，避免了昂贵的忙等待
始终使用 wait() 的谓词形式来处理虚假唤醒和丢失唤醒
生产者-消费者模式使用两个条件变量：「未满」和「非空」
notify_one() 唤醒一个等待者；notify_all() 唤醒所有等待者 ── 根据场景选择正确的方法
等待者和通知者都必须使用同一个互斥锁来保护共享条件

测验 — 检验你的知识

(20 经验值)

1. 为什么你必须始终在 `condition_variable::wait()` 中使用谓词？

2. 在有界的生产者-消费者队列中，当队列满时生产者应该怎么做？

3. `std::condition_variable::wait()` 需要哪种锁类型？

上一课

互斥锁、锁与临界区

下一课

std::async、std::future 与 std::promise