一、从线程安全队列到生产者消费者模型

上一篇中使用：

std::mutex

配合：

std::deque

实现了一个简单的线程安全队列。
它解决的核心问题是：

多个线程同时操作队列 ↓ 使用 mutex 进行保护 ↓ 同一时刻只有一个线程能够修改队列 ↓ 避免数据竞争

但是这种简单队列还有一个问题。
假设消费者不断从队列中取数据：

int value; while (queue.next(value)) { // 处理数据 }

如果消费者执行到这里时队列恰好为空：

消费者 ↓ 检查队列 ↓ 队列为空 ↓ next() 返回 false ↓ 消费者结束

可是生产者可能只是暂时还没有来得及添加数据。
例如：

消费者启动 ↓ 发现队列为空 ↓ 直接退出 过了一会 生产者 ↓ 添加任务

这样新加入的数据就没有消费者处理了。
一种简单粗暴的方法是让消费者一直检查：

while (true) { if (!queue.empty()) { // 获取任务 } }

但是这种方式会导致线程不停循环：

检查队列 ↓ 没有数据 ↓ 继续检查 ↓ 还是没有 ↓ 继续检查……

即使没有任何任务，CPU 也一直在工作，这种情况通常称为忙等待（Busy Waiting）。
更合理的方式应该是：

队列有数据 ↓ 消费者取数据 队列为空 ↓ 消费者睡眠等待 生产者加入数据 ↓ 通知消费者 消费者被唤醒 ↓ 继续取数据

这就是std::condition_variable的主要作用。
因此一个比较完整的生产者消费者队列通常需要：

std::queue<T> _queue; // 保存数据 std::mutex _queueLock; // 保护队列 std::condition_variable _condition; // 控制线程等待和唤醒

三者之间可以简单理解为：

┌──────────────┐ 生产者 ──Push()──→ │ queue │ ──Pop()──→ 消费者 └──────────────┘ ↑ mutex ↑ condition_variable 等待 / 唤醒线程

mutex负责解决：

多个线程能不能同时修改队列？
而condition_variable解决的是：
队列没有数据时，消费者应该怎么办？

二、ProducerConsumerQueue 的基本结构

先来看一个生产者消费者队列需要哪些成员变量：

#include <condition_variable> #include <mutex> #include <queue> #include <atomic> template<typename T> class ProducerConsumerQueue { private: std::mutex _queueLock; // 保护队列 std::queue<T> _queue; // 保存生产者产生的数据 std::condition_variable _condition; // 控制消费者等待和唤醒 std::atomic<bool> _shutdown; // 队列是否停止工作 public: ProducerConsumerQueue() : _shutdown(false) {} };

其中_queue是真正保存数据的地方：

std::queue<T> _queue;

例如：

ProducerConsumerQueue<int> queue;

内部就相当于：

std::queue<int> _queue;

队列遵循 FIFO：

First In First Out 先进先出

例如：

push(10) push(20) push(30) 队头 队尾 ↓ ↓ 10 → 20 → 30

消费者第一次取出：

10

第二次取出：

20

然后是：

30

第二个成员：

std::mutex _queueLock;

用来保护_queue。
生产者执行：

_queue.push(value);

消费者执行：

_queue.pop();

这些操作不能被多个线程同时执行，因此需要_queueLock。
第三个成员：

std::condition_variable _condition;

它负责消费者线程的：

等待 + 唤醒

最后：

std::atomic<bool> _shutdown;

用于表示整个队列是否准备结束。
初始状态：

_shutdown = false;

表示：

队列正常运行

调用取消操作以后：

_shutdown = true;

表示：

队列准备停止

这几个成员组合起来，就形成了一个比较典型的生产者消费者队列：

ProducerConsumerQueue ┌──────────────────────────────┐ │ │ │ std::queue<T> _queue │ │ ↑ │ │ std::mutex _queueLock │ │ │ │ condition_variable │ │ 等待 / 唤醒 │ │ │ │ atomic<bool> _shutdown │ │ 控制退出 │ │ │ └──────────────────────────────┘

三、Push：生产者添加数据并唤醒消费者

生产者向队列添加数据可以实现为：

void Push(const T& value) { std::lock_guard<std::mutex> lock(_queueLock); // 获取互斥锁 _queue.push(value); // 数据加入队列 _condition.notify_one(); // 唤醒一个等待中的消费者 }

首先：

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

给队列加锁。
然后：

_queue.push(value);

将数据加入队列。
例如：

queue.Push(10);

队列变成：

┌────┐ │ 10 │ └────┘

再次：

queue.Push(20);

变成：

10 → 20

真正值得注意的是最后一句：

_condition.notify_one();

它的意思不是“通知队列”，而是：

唤醒一个正在_condition上等待的线程。
假设现在消费者因为队列为空进入等待：

消费者线程 ↓ 队列为空 ↓ condition_variable ↓ 进入等待状态

此时生产者：

queue.Push(100);

执行过程：

生产者获得锁 ↓ push(100) ↓ 队列中出现数据 ↓ notify_one() ↓ 唤醒一个消费者

消费者被唤醒以后继续尝试获取数据：

queue ↑ 生产者 ──push()──→ [100] │ notify_one() ↓ 消费者 ↓ 唤醒

这里有两个比较容易混淆的函数：

notify_one(); notify_all();

notify_one()：

唤醒一个等待线程。
notify_all()：
唤醒所有等待线程。
生产者每次通常只添加一个任务，因此：

_condition.notify_one();

一般就足够了。
例如三个消费者都在等待：

消费者1：等待 消费者2：等待 消费者3：等待

生产者加入一个任务：

queue.Push(100);

调用：

notify_one();

可能只有消费者2被唤醒：

消费者1：继续等待 消费者2：被唤醒 → 获取100 消费者3：继续等待

这样避免了明明只有一个任务，却把所有消费者全部叫醒。
另外，有些代码中可能会看到：

_queue.push(std::move(value));

如果Push()的参数本身是：

const T& value

那么这里的std::move(value)实际得到的是const T&&，很多类型最终仍然会发生拷贝，而不是真正的移动。
如果希望真正支持移动，可以额外提供：

void Push(T&& value) { std::lock_guard<std::mutex> lock(_queueLock); _queue.push(std::move(value)); _condition.notify_one(); }

不过理解生产者消费者模型时，暂时把重点放在：

push + notify_one

这一组操作上即可。

四、Pop 与 WaitAndPop：普通取数据和阻塞等待的区别

消费者取数据可以有两种方式。
第一种是普通的Pop()：

bool Pop(T& value) { std::lock_guard<std::mutex> lock(_queueLock); if (_queue.empty() || _shutdown) return false; value = _queue.front(); _queue.pop(); return true; }

首先获取锁：

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

然后判断：

if (_queue.empty() || _shutdown) return false;

只要：

队列为空

或者：

队列已经停止

就不继续取数据。
如果存在数据：

value = _queue.front(); _queue.pop();

先获取队头：

value = _queue.front();

再删除队头：

_queue.pop();

例如：

10 → 20 → 30

执行：

queue.Pop(value);

之后：

value = 10 队列： 20 → 30

这种Pop()有一个特点：

没数据就直接返回，不会等待。
真正体现生产者消费者模型的是WaitAndPop()：

void WaitAndPop(T& value) { std::unique_lock<std::mutex> lock(_queueLock); while (_queue.empty() && !_shutdown) _condition.wait(lock); if (_queue.empty() || _shutdown) return; value = _queue.front(); _queue.pop(); }

这里首先出现：

std::unique_lock<std::mutex> lock(_queueLock);

为什么这里不用前面的：

std::lock_guard<std::mutex>

而要使用：

std::unique_lock<std::mutex>

因为后面需要执行：

_condition.wait(lock);

condition_variable::wait()在等待过程中需要：

释放 mutex ↓ 线程睡眠 ↓ 收到通知 ↓ 重新获取 mutex

这种“中途释放锁、醒来以后重新加锁”的操作需要unique_lock配合完成。
最关键的代码就是：

while (_queue.empty() && !_shutdown) _condition.wait(lock);

可以先把条件翻译成人话：

队列为空 并且 程序还没有关闭

那么：

_condition.wait(lock);

消费者进入等待。
假设当前：

_queue.empty() == true _shutdown == false

那么消费者执行：

进入 wait ↓ 释放 mutex ↓ 消费者线程睡眠

为什么wait()必须把锁释放？
因为如果消费者睡着以后还一直拿着_queueLock：

消费者拿着锁睡觉 ↓ 生产者 Push() ↓ 想获得 _queueLock ↓ 拿不到 ↓ 无法添加数据

这样生产者永远无法把数据放进去，消费者也永远等不到数据。
所以：

_condition.wait(lock);

内部非常重要的一件事就是：

等待时自动释放互斥锁。
完整流程可以理解为：

消费者执行 WaitAndPop() ↓ 获得 _queueLock ↓ 检查队列 ↓ 队列为空 ↓ wait(lock) ↓ 释放 _queueLock ↓ 消费者睡眠 ↓ -------------------------- ↓ 生产者获得 _queueLock ↓ push() ↓ notify_one() ↓ -------------------------- ↓ 消费者被唤醒 ↓ 重新竞争 _queueLock ↓ 获得锁 ↓ 重新检查条件 ↓ front() ↓ pop()

这里还有一个非常重要的细节：

while (_queue.empty() && !_shutdown)

为什么是while，而不是：

if (_queue.empty() && !_shutdown)

因为条件变量存在虚假唤醒（Spurious Wakeup）。
线程被唤醒，并不意味着：

队列一定有数据

因此醒来之后还必须重新检查：

_queue.empty()

所以通常应该写成：

while (条件不满足) condition.wait(lock);

C++ 还提供了更加常见的谓词版本：

_condition.wait(lock, [this]() { return !_queue.empty() || _shutdown; });

这句可以理解为：

一直等待，直到队列中有数据，或者整个队列准备退出。
实际上它内部的思想和：

while (_queue.empty() && !_shutdown) _condition.wait(lock);

基本一致。
使用谓词以后，WaitAndPop()可以写成：

void WaitAndPop(T& value) { std::unique_lock<std::mutex> lock(_queueLock); _condition.wait(lock, [this]() { return !_queue.empty() || _shutdown; }); if (_queue.empty() || _shutdown) return; value = _queue.front(); _queue.pop(); }

这也是实际编写条件变量代码时很常见的一种写法。

五、Cancel：如何让正在等待的消费者安全退出

最后还需要解决一个问题：

如果程序准备结束，但是消费者还在wait()里面睡觉怎么办？
假设现在有三个消费者：

消费者1 → wait() 消费者2 → wait() 消费者3 → wait()

此时程序准备退出。
如果什么都不做，这些线程可能仍然处于等待状态。
因此需要提供：

void Cancel()

通知整个队列结束。
一个基本实现为：

void Cancel() { { std::lock_guard<std::mutex> lock(_queueLock); _shutdown = true; } _condition.notify_all(); }

首先：

_shutdown = true;

告诉消费者：

队列已经停止工作

然后：

_condition.notify_all();

把所有正在等待的消费者全部唤醒。
为什么这里不是：

notify_one();

因为程序准备退出时，希望：

消费者1 消费者2 消费者3

全部醒来并退出，而不是只叫醒其中一个。
执行过程：

Cancel() ↓ _shutdown = true ↓ notify_all() ↓ ┌─────────┬─────────┬─────────┐ ↓ ↓ ↓ 消费者1 消费者2 消费者3 唤醒 唤醒 唤醒 ↓ ↓ ↓ 发现 shutdown == true ↓ ↓ ↓ 退出 退出 退出

把前面的内容组合起来，一个简单的生产者消费者队列可以写成：

#include <atomic> #include <condition_variable> #include <mutex> #include <queue> template<typename T> class ProducerConsumerQueue { private: std::mutex _queueLock; // 保护队列 std::queue<T> _queue; // 保存数据 std::condition_variable _condition; // 控制等待和唤醒 std::atomic<bool> _shutdown; // 是否停止 public: ProducerConsumerQueue() : _shutdown(false) {} // 生产者添加数据 void Push(const T& value) { { std::lock_guard<std::mutex> lock(_queueLock); _queue.push(value); } _condition.notify_one(); } // 普通取数据：没有数据直接返回 false bool Pop(T& value) { std::lock_guard<std::mutex> lock(_queueLock); if (_queue.empty() || _shutdown) return false; value = _queue.front(); _queue.pop(); return true; } // 阻塞取数据：没有数据就等待 bool WaitAndPop(T& value) { std::unique_lock<std::mutex> lock(_queueLock); _condition.wait(lock, [this]() { return !_queue.empty() || _shutdown; }); if (_shutdown && _queue.empty()) return false; value = _queue.front(); _queue.pop(); return true; } // 停止队列并唤醒所有消费者 void Cancel() { { std::lock_guard<std::mutex> lock(_queueLock); _shutdown = true; } _condition.notify_all(); } };

然后可以创建一个简单的生产者消费者程序：

#include <iostream> #include <thread> int main() { ProducerConsumerQueue<int> queue; // 消费者：没有数据时会阻塞等待 std::thread consumer([&]() { int value; while (queue.WaitAndPop(value)) { std::cout << "consumer : " << value << std::endl; } }); // 生产者：不断向队列中添加数据 std::thread producer([&]() { queue.Push(10); queue.Push(20); queue.Push(30); queue.Push(40); queue.Cancel(); }); producer.join(); consumer.join(); return 0; }

整个程序的工作流程就是：

消费者启动 ↓ WaitAndPop() ↓ 队列为空 ↓ wait() 睡眠 ↓ ════════════════════ ↓ 生产者 Push(10) ↓ notify_one() ↓ ════════════════════ ↓ 消费者被唤醒 ↓ 取出10 ↓ 继续 WaitAndPop()

因此相比上一篇简单的：

deque + mutex

这一篇又增加了：

condition_variable

最终形成：

mutex ↓ 生产者 → 线程安全队列 → 消费者 │ ↑ └──── notify_one ───────┘ │ wait()

这一部分最需要掌握的其实就是下面几组对应关系：

std::mutex // 保护共享队列 std::lock_guard<std::mutex> // 简单加锁 std::unique_lock<std::mutex> // 配合 condition_variable 使用

以及：

_condition.wait(lock); // 消费者等待 _condition.notify_one(); // 唤醒一个消费者 _condition.notify_all(); // 唤醒所有消费者

最终把整个生产者消费者模型概括成一句话就是：

生产者负责向队列中放数据，消费者负责从队列中取数据；mutex 保证队列访问安全，condition_variable 负责在没有数据时让消费者休眠，并在新数据到来后将其唤醒。
理解这一套流程之后，再继续学习线程池会非常自然，因为线程池中的：

提交任务 ↓ 任务队列 ↓ 工作线程等待 ↓ 有任务后唤醒 ↓ 执行任务

本质上就是生产者消费者模型的一种典型应用。

0voice · GitHub