【C++】条件变量：wait、notify_one 与 notify_all 到底怎么工作的

前面介绍互斥锁时，我们解决的是：

多个线程同时访问共享数据时，如何保证同一时刻只有一个线程进入临界区。

但是互斥锁还有一个解决不了的问题。

假设线程池中的工作线程需要不断检查任务队列：

while (true) { std::lock_guard<std::mutex> lock(task_mutex); if (!tasks.empty()) { // 取任务 } }

如果队列一直为空，线程就会不停循环：

检查队列 ↓ 没有任务 ↓ 继续检查 ↓ 还是没有 ↓ 继续检查……

这样会白白消耗 CPU。

我们真正希望的是：

没有任务 ↓ 线程睡眠 ↓ 其他线程放入任务 ↓ 唤醒工作线程 ↓ 继续处理任务

这就是std::condition_variable条件变量的作用。

一、condition_variable 到底解决什么问题

先定义：

#include <condition_variable> #include <mutex> #include <queue> std::mutex mutex; std::condition_variable condition; std::queue<int> tasks;

工作线程可以这样等待任务：

void Worker() { std::unique_lock<std::mutex> lock(mutex); condition.wait(lock, []() { return !tasks.empty(); }); int task = tasks.front(); tasks.pop(); }

生产者线程加入任务：

void AddTask(int task) { { std::lock_guard<std::mutex> lock(mutex); tasks.push(task); } condition.notify_one(); }

整个过程是：

Worker线程 ↓ 检查tasks.empty() ↓ 为空 ↓ condition.wait() ↓ 释放mutex并睡眠 AddTask线程 ↓ 获得mutex ↓ tasks.push() ↓ 释放mutex ↓ notify_one() ↓ 唤醒Worker

这里一定要注意：

条件变量本身并不保存“任务”。

真正的数据还是：

std::queue<int> tasks;

条件变量只是负责：

等待某个条件成立 + 通知等待线程重新检查条件

所以：

condition.notify_one();

并不是把任务传递给工作线程。

真正传递的数据仍然放在：

tasks

里面。

二、wait(lock, predicate) 到底是怎么执行的

线程池中最常见的一句代码就是：

condition.wait(lock, [this]() { return stop_ || !tasks_.empty(); });

这句代码第一次看确实很容易懵。

先把 Lambda 单独拿出来：

[this]() { return stop_ || !tasks_.empty(); }

它本质上就是一个函数。

可以简单理解成：

bool Check() { return stop_ || !tasks_.empty(); }

所以：

condition.wait(lock, [this]() { return stop_ || !tasks_.empty(); });

逻辑上类似：

condition.wait(lock, Check);

而带谓词版本的wait()可以近似理解成：

while (!(stop_ || !tasks_.empty())) { condition.wait(lock); }

也就是说，它会不断检查：

stop_ || !tasks_.empty()

情况一：返回 true

例如：

stop_ = false tasks_不为空

那么：

stop_ || !tasks_.empty()

结果就是：

false || true = true

这时候：

condition.wait(lock, predicate);

不会进入等待，直接继续往下执行。

可以理解成：

condition.wait(lock, true);

虽然真正的函数并不是这样传参，但从理解执行结果的角度可以这么看。

情况二：返回 false

例如：

stop_ = false tasks_为空

得到：

false || false = false

这时候线程就需要等待。

但wait()做的事情不是简单的“停在这里”。

它实际上会完成：

当前线程持有mutex ↓ predicate返回false ↓ wait自动释放mutex ↓ 当前线程进入睡眠 ↓ 其他线程notify ↓ 当前线程被唤醒 ↓ 重新获取mutex ↓ 再次检查predicate

这里最重要的一点是：

wait()在睡眠期间会释放互斥锁。

否则就会出问题。

假设工作线程睡眠时还一直拿着：

mutex

那么生产者想执行：

tasks.push(task);

也需要这把锁：

std::lock_guard<std::mutex> lock(mutex);

结果就变成：

Worker： 拿着mutex睡觉 Producer： 想拿mutex加入任务 ↓ 拿不到 Worker： 等Producer添加任务 ↓ Producer又因为拿不到mutex无法添加

程序就卡住了。

所以wait()必须：

睡眠前释放锁 唤醒后重新获得锁

三、为什么必须使用 unique_lock

因此我们就能理解一个非常经典的问题：

为什么条件变量要写：

std::unique_lock<std::mutex> lock(mutex); condition.wait(lock);

而不能写：

std::lock_guard<std::mutex> lock(mutex); condition.wait(lock);

原因就在于：

condition_variable::wait()

内部需要控制这把锁：

unlock ↓ 睡眠 ↓ lock

而lock_guard的设计非常简单：

构造 → 加锁 析构 → 解锁

它不允许中途主动：

lock.unlock(); lock.lock();

而unique_lock可以：

std::unique_lock<std::mutex> lock(mutex); lock.unlock(); lock.lock();

所以条件变量需要unique_lock。

可以把它理解成：

lock_guard： “我拿到锁以后一直拿到作用域结束。” unique_lock： “我负责管理这把锁，但中途允许释放和重新获取。”

而wait()恰好就需要第二种能力。

例如：

std::unique_lock<std::mutex> lock(mutex); condition.wait(lock, []() { return !tasks.empty(); });

内部大致相当于：

检查tasks ↓ 没有任务 ↓ lock.unlock() ↓ 线程睡眠 ↓ 被notify唤醒 ↓ lock.lock() ↓ 再次检查tasks

因此在线程池、阻塞队列、生产者消费者模型中，unique_lock + condition_variable经常一起出现。

四、notify_one、notify_all 与虚假唤醒

当生产者添加新任务以后，需要通知正在等待的线程：

condition.notify_one();

或者：

condition.notify_all();

notify_one

condition.notify_one();

表示：

唤醒一个正在等待这个条件变量的线程。

假设有 4 个工作线程：

Worker1：等待 Worker2：等待 Worker3：等待 Worker4：等待

加入一个任务：

tasks.push(task); condition.notify_one();

只需要唤醒其中一个线程处理任务即可。

因此线程池添加普通任务时，一般使用：

condition.notify_one();

notify_all

condition.notify_all();

表示：

唤醒所有正在等待的线程。

例如线程池准备关闭：

stop_ = true; condition.notify_all();

此时：

Worker1 Worker2 Worker3 Worker4

都需要知道：

线程池准备退出

所以应该全部唤醒。

典型写法：

{ std::lock_guard<std::mutex> lock(tasks_mutex_); stop_ = true; } condition_.notify_all();

所有线程醒来以后重新检查：

return stop_ || !tasks_.empty();

因为：

stop_ == true

所以谓词返回：

true

线程不再继续等待，而是进入后面的退出逻辑。

为什么唤醒后还要重新检查条件

因为条件变量存在一个重要现象：

Spurious Wakeup 虚假唤醒

也就是说：

即使没有调用notify_one()或notify_all()，等待线程也有可能被唤醒。

所以不能这样写：

condition.wait(lock); // 醒了就默认一定有任务 int task = tasks.front();

因为线程醒来以后：

tasks

可能仍然为空。

传统写法应该是：

while (tasks.empty()) { condition.wait(lock); }

也就是：

醒来 ↓ 重新检查条件 ↓ 条件不满足 ↓ 继续睡眠

而现代 C++ 更推荐：

condition.wait(lock, []() { return !tasks.empty(); });

因为带谓词的版本已经帮我们处理了：

while (!predicate()) { wait(lock); }

所以线程池中的：

condition_.wait(lock, [this]() { return stop_ || !tasks_.empty(); });

写法实际上非常标准。

五、完整生产者—消费者示例

最后用一个简单任务队列，把整个流程串起来。

#include <condition_variable> #include <iostream> #include <mutex> #include <queue> #include <thread> std::queue<int> tasks; std::mutex tasks_mutex; std::condition_variable condition; bool stop = false;

消费者线程：

void Worker() { while (true) { std::unique_lock<std::mutex> lock(tasks_mutex); condition.wait(lock, []() { return stop || !tasks.empty(); }); if (stop && tasks.empty()) { return; } int task = tasks.front(); tasks.pop(); lock.unlock(); std::cout << "处理任务：" << task << '\n'; } }

生产者：

void AddTask(int task) { { std::lock_guard<std::mutex> lock(tasks_mutex); tasks.push(task); } condition.notify_one(); }

主函数：

int main() { std::thread worker1(Worker); std::thread worker2(Worker); AddTask(100); AddTask(200); AddTask(300); { std::lock_guard<std::mutex> lock(tasks_mutex); stop = true; } condition.notify_all(); worker1.join(); worker2.join(); return 0; }

这里有一个值得注意的地方：

int task = tasks.front(); tasks.pop(); lock.unlock(); std::cout << "处理任务：" << task << '\n';

为什么取出任务后主动：

lock.unlock();

因为互斥锁只需要保护：

访问tasks队列

真正执行任务不应该一直占着锁。

否则：

Worker1拿到锁 ↓ 取出任务 ↓ 一直拿着锁执行任务 ↓ Worker2无法访问任务队列

线程池就会失去并发执行的意义。

所以正确思路一般是：

加锁 ↓ 访问共享任务队列 ↓ 取出任务 ↓ 解锁 ↓ 真正执行任务

这也是unique_lock灵活性的一个典型体现。

整个生产者—消费者模型可以总结为：

生产者 ↓ 获得mutex ↓ 加入任务 ↓ 释放mutex ↓ notify_one 消费者 ↓ 获得mutex ↓ 检查条件 ↓ 没有任务 ↓ wait释放mutex并睡眠 ↓ 被唤醒 ↓ 重新获得mutex ↓ 再次检查条件 ↓ 取出任务 ↓ 释放mutex ↓ 执行任务

这一篇最需要掌握的其实就是下面几句话：

condition_variable用于“等待某个条件成立”； wait睡眠时会自动释放mutex， 醒来后会重新获得mutex； 正因为需要中途unlock和lock， 所以wait通常配合unique_lock； predicate返回true时不等待， 返回false时进入等待； 线程醒来以后必须重新检查条件， 因此推荐使用带predicate的wait； notify_one通常唤醒一个工作线程； notify_all通常用于退出、状态变化等 需要所有等待线程都重新检查条件的场景。

如果把线程池中的这一句真正理解：

condition_.wait(lock, [this]() { return stop_ || !tasks_.empty(); });

那么condition_variable最核心的部分基本就掌握了。

0voice · GitHub