一、为什么需要线程安全队列

在 C++ 中，我们经常使用std::queue、std::deque等容器保存数据。例如：

std::deque<int> queue; queue.push_back(1); queue.push_back(2);

在单线程程序中这样使用没有什么问题，但是如果多个线程同时操作同一个队列，就可能产生数据竞争。
例如现在有两个线程：

std::thread t1([&]() { queue.push_back(1); }); std::thread t2([&]() { queue.push_back(2); });

两个线程可能同时进入push_back()，共同修改queue内部的数据结构。
需要注意的是：

std::deque std::queue std::vector

这些 STL 容器本身并不会自动帮我们解决多个线程同时读写的问题。
因此最直接的解决方案就是：

普通队列 ↓ 加入 mutex ↓ 每次操作队列之前先加锁 ↓ 操作结束后释放锁 ↓ 线程安全队列

例如：

std::mutex mutex; std::deque<int> queue;

线程操作队列的时候：

mutex.lock(); queue.push_back(1); mutex.unlock();

这样同一时刻只能有一个线程操作队列。
假设线程1已经获得锁：

线程1：获得 mutex ↓ push_back() ↓ 释放锁

线程2此时也想操作队列：

线程2：尝试获得 mutex ↓ 获取失败 ↓ 等待 ↓ 线程1释放 mutex ↓ 线程2获得 mutex

这就是线程安全队列最基础的实现思路。

二、LockedQueue 的基本结构

可以把线程安全队列封装成一个模板类：

#include <deque> #include <mutex> template<class T, typename StorageType = std::deque<T>> class LockedQueue { private: std::mutex _lock; // 保护队列的互斥锁 StorageType _queue; // 真正保存数据的容器 bool _canceled; // 队列是否已经取消 public: LockedQueue() : _canceled(false) {} };

这里最核心的其实只有两个成员：

std::mutex _lock; StorageType _queue;

其中_queue负责保存数据，而_lock负责保护_queue。
模板：

template<class T, typename StorageType = std::deque<T>>

表示这个队列保存的数据类型由T决定，同时默认使用：

std::deque<T>

作为底层容器。
例如：

LockedQueue<int> queue;

此时内部实际上可以理解成：

std::deque<int> _queue;

如果创建：

LockedQueue<std::string> queue;

那么内部就是：

std::deque<std::string> _queue;

因此LockedQueue可以保存不同类型的数据。
整个结构可以简单理解成：

LockedQueue<int> ┌─────────────────────┐ │ mutex _lock │ │ ↓ │ │ ┌─────────────────┐ │ │ │ 1 │ 2 │ 3 │ 4 │ │ │ └─────────────────┘ │ │ _queue │ └─────────────────────┘

只要所有对_queue的操作都经过_lock，就能够避免多个线程同时修改队列。

三、实现线程安全的入队操作

首先实现最基本的入队操作：

void add(const T& item) { _lock.lock(); // 操作队列之前先加锁 _queue.push_back(item); // 将数据加入队尾 _lock.unlock(); // 操作完成后释放锁 }

例如：

LockedQueue<int> queue; queue.add(10); queue.add(20); queue.add(30);

队列中的数据就是：

队头 队尾 ↓ ↓ 10 → 20 → 30

其中：

_queue.push_back(item);

负责真正添加数据。
而：

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

负责保证同一时刻只能有一个线程执行这段操作。
假设两个线程同时调用：

queue.add(1); queue.add(2);

实际执行过程可能是：

线程1 ↓ 获得锁 ↓ push_back(1) ↓ 释放锁 线程2 ↓ 获得锁 ↓ push_back(2) ↓ 释放锁

因此两个线程不会同时修改_queue。
不过在实际 C++ 开发中，一般更加推荐使用 RAII 的方式管理锁：

void add(const T& item) { std::lock_guard<std::mutex> lock(_lock); _queue.push_back(item); }

这里：

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

创建时自动执行加锁，函数结束时自动释放锁。
可以理解为：

进入函数 ↓ 创建 lock_guard ↓ mutex.lock() ↓ 操作 queue ↓ 函数结束 ↓ lock_guard 析构 ↓ mutex.unlock()

与手动：

_lock.lock(); _queue.push_back(item); _lock.unlock();

相比，lock_guard最大的好处就是不用担心中途return或发生异常导致忘记释放锁。
所以简单的临界区代码通常推荐：

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

四、实现线程安全的出队操作

只有入队还不够，还需要能够安全地从队列中取出数据。
可以实现一个next()：

bool next(T& result) { std::lock_guard<std::mutex> lock(_lock); if (_queue.empty()) return false; result = _queue.front(); // 取出队头元素 _queue.pop_front(); // 删除队头元素 return true; }

这里的逻辑非常重要。
首先：

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

获得互斥锁。
然后判断：

if (_queue.empty()) return false;

如果队列已经没有数据，就直接返回false。
如果有数据：

result = _queue.front();

先把队头元素保存到result中。
然后：

_queue.pop_front();

删除队头元素。
最后：

return true;

表示成功取到了数据。
例如队列：

10 → 20 → 30

执行：

int value; queue.next(value);

首先：

value = _queue.front();

此时：

value = 10

然后：

_queue.pop_front();

队列变成：

20 → 30

因此next()实际上完成了：

获取队头元素 + 删除队头元素

为什么必须把：

empty() front() pop_front()

全部放在同一把锁里面？
因为如果分开加锁，就可能出现这种情况：

线程1：检查队列不为空 ↓ 释放锁 线程2：取走最后一个元素 线程1：继续 front()

此时线程1再访问队头就会出现问题。
所以：

if (_queue.empty()) return false; result = _queue.front(); _queue.pop_front();

应该看成一个完整操作，整个过程都由同一把锁保护。
这也是多线程编程中非常重要的一点：

不仅仅是单独保护某一行代码，更重要的是保护一个完整的逻辑操作。

有了next()之后，就可以这样不断读取队列：

int value; while (queue.next(value)) { std::cout << value << std::endl; }

只要队列里面还有数据：

queue.next(value)

就返回true。
当队列为空时返回false，循环结束。

五、多线程测试 LockedQueue

现在把前面的内容组合起来，实现一个简单的线程安全队列：

#include <deque> #include <mutex> template<class T, typename StorageType = std::deque<T>> class LockedQueue { private: std::mutex _lock; // 保护队列 StorageType _queue; // 保存数据 bool _canceled; // 取消标志 public: LockedQueue() : _canceled(false) {} // 添加元素 void add(const T& item) { std::lock_guard<std::mutex> lock(_lock); _queue.push_back(item); } // 获取并删除队头元素 bool next(T& result) { std::lock_guard<std::mutex> lock(_lock); if (_queue.empty()) return false; result = _queue.front(); _queue.pop_front(); return true; } // 判断队列是否为空 bool empty() { std::lock_guard<std::mutex> lock(_lock); return _queue.empty(); } // 取消队列 void cancel() { std::lock_guard<std::mutex> lock(_lock); _canceled = true; } // 判断队列是否已经取消 bool cancelled() { std::lock_guard<std::mutex> lock(_lock); return _canceled; } };

接下来创建两个生产线程向队列添加数据，再创建两个线程读取数据：

#include <iostream> #include <thread> #include "LockedQueue.h" int main() { LockedQueue<int> queue; // 生产者线程1 std::thread producer1([&]() { queue.add(1); queue.add(2); queue.add(3); queue.add(4); }); // 生产者线程2 std::thread producer2([&]() { queue.add(5); queue.add(6); queue.add(7); queue.add(8); }); // 消费者线程1 std::thread consumer1([&]() { int value; while (queue.next(value)) { std::cout << std::this_thread::get_id() << " : pop " << value << std::endl; } }); // 消费者线程2 std::thread consumer2([&]() { int value; while (queue.next(value)) { std::cout << std::this_thread::get_id() << " : pop " << value << std::endl; } }); producer1.join(); producer2.join(); consumer1.join(); consumer2.join(); return 0; }

这里一共创建了四个线程：

producer1 ──→ 添加 1 2 3 4 ──┐ │ ↓ LockedQueue ↑ │ producer2 ──→ 添加 5 6 7 8 ──┘ ↓ ┌────────┴────────┐ ↓ ↓ consumer1 consumer2 取出数据 取出数据

两个生产者线程：

producer1 producer2

不断调用：

queue.add();

向队列中添加数据。
两个消费者线程：

consumer1 consumer2

不断调用：

queue.next();

获取数据。
因为add()和next()内部都使用了同一把：

std::mutex _lock;

所以无论有多少线程操作这个队列，同一时刻都只能有一个线程真正访问_queue。
需要注意一个细节，这个简单版本还有一个明显的问题：

while (queue.next(value))

当消费者执行next()时，如果队列刚好为空，它会直接返回false。
也就是说：

消费者发现队列为空 ↓ next() 返回 false ↓ 消费者线程结束

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

消费者线程启动 ↓ 发现队列为空 ↓ 退出 ↓ 生产者线程 ↓ 添加数据

这样生产者后面加入的数据就没人处理了。
因此当前这个LockedQueue解决的是：

多个线程同时操作队列时的数据安全问题。
但是它还没有解决：
队列为空时，消费者应该如何等待生产者的问题。
如果让消费者不断这样检查：

while (true) { if (queue.next(value)) { // 处理数据 } }

又会产生大量无意义的循环，占用 CPU。
更好的办法是：

队列有数据 ↓ 消费者正常取数据 队列没数据 ↓ 消费者进入等待状态 生产者添加数据 ↓ 通知消费者 ↓ 消费者被唤醒 ↓ 继续取数据

而实现这个过程就需要 C++ 多线程中的另一个重要工具：

std::condition_variable

这也会进一步形成经典的：

生产者 ——→ 线程安全队列 ——→ 消费者

生产者消费者模型。

总结

这一部分实现的LockedQueue本质上并不复杂，它的核心结构就是：

std::deque + std::mutex = 简单线程安全队列

其中最重要的几个操作是：

_queue.push_back(item); // 入队 _queue.front(); // 获取队头 _queue.pop_front(); // 删除队头

而所有这些操作外面都通过：

std::mutex

进行保护。
其中：

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

利用 RAII 自动完成加锁和解锁，是简单临界区中非常常见的写法。
到这里解决的是：

多个线程同时操作队列 ↓ mutex 保证同一时间只有一个线程访问 ↓ 避免数据竞争

但是还存在：

队列为空 ↓ 消费者应该怎么办？

这个问题。
因此在线程安全队列的基础上，下一步就可以继续加入：

std::condition_variable

让消费者在队列为空时进入等待，在生产者加入数据之后再将消费者唤醒，从而实现一个更加完整的生产者消费者队列。

0voice · GitHub