一、从线程安全队列到生产者消费者模型
上一篇中使用：
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
