前言消息队列要解决的是生产者与消费者之间的解耦和速度不匹配问题生产线程产生任务的速度可能忽快忽慢消费线程处理任务的速度也有波动中间放一个缓冲区双方就不必互相等待。在单进程内这个缓冲区通常就是一个受互斥量保护的容器加上条件变量做等待与唤醒。初学者最常见的误解有三个。第一以为std::queue本身是线程安全的直接多线程读写实际上标准库容器都不是线程安全的并发读写同一容器是数据竞争属于未定义行为undefined behavior, UB。第二以为condition_variable::wait醒过来就说明条件成立实际上存在虚假唤醒spurious wakeup必须用谓词循环检查。第三只考虑正常收发不考虑关闭流程结果消费者永远阻塞在wait上程序无法退出。本文从有界缓冲区的语义讲起说明互斥量与两个条件变量如何配合给出一份可以直接放进项目的模板实现再讨论关闭顺序、异常安全和常见写法错误。文中代码以 C17 为基准只使用标准库。一、有界队列的语义边界一个实用的阻塞队列至少要明确四件事容量、阻塞策略、关闭策略、返回值语义。容量决定了背压backpressure。无界队列在生产者远快于消费者时会持续增长最终耗尽内存有界队列则让生产者在队列满时阻塞把压力传回上游。绝大多数任务分发场景都应该选择有界队列容量由内存预算和处理延迟共同决定。阻塞策略有两种常见组合push在满时等待、pop在空时等待这是标准的阻塞队列也可以提供非阻塞的try_push/try_pop立即返回成功或失败。两者可以并存非阻塞版本在超时、轮询、避免死锁的场景下有用。关闭策略是最容易被忽略的部分。队列关闭后通常的约定是不再接受新的push已经入队的元素仍然可以被pop取走当队列排空后pop返回一个表示结束的值。这样消费者可以自然退出而不是靠一个全局标志轮询。返回值语义要能区分三种情况取到了值、队列已关闭且为空、操作失败。用std::optionalT表达前两种最直接有值表示取到std::nullopt表示结束。如果T本身可以是空也不受影响因为optional表达的是有没有值而不是值是否为空。二、互斥量与条件变量如何配合队列的内部状态由三部分组成元素容器、关闭标志、容量。这三者被多个线程同时访问必须用同一把互斥量保护。任何读或写这些状态的地方都要持锁包括empty()这种看起来无害的查询。条件变量本身不存储状态它只是一个等待队列。因此判断能不能继续的依据必须放在共享状态里而不是指望条件变量的返回值。标准的写法是带谓词的wait// 适用于 C17not_empty_.wait(lock, [this] { return closed_ || !queue_.empty(); });这个重载等价于在循环里调用无谓词版本每次被唤醒都重新求值谓词谓词为假就继续睡。谓词在持有锁的状态下求值所以读queue_和closed_是安全的。不要写成if (!queue_.empty()) wait(lock);那是典型的丢失唤醒lost wakeup写法。为什么需要两个条件变量因为生产者等待不满、消费者等待不空条件不同。如果共用一个条件变量并用notify_all每次状态变化都会唤醒所有线程其中大部分醒来后发现条件不满足又睡回去形成惊群。用两个条件变量加notify_one可以让唤醒更有针对性。不过notify_one并不保证唤醒的是最该醒的那个线程标准只保证至少唤醒一个等待者。即使如此正确性仍由谓词循环保证被错误唤醒的线程重新检查条件不满足就继续等。这也是为什么谓词不能省。唤醒时机也有讲究。在持锁期间调用notify_one是合法的但被唤醒的线程要立刻去抢锁而锁还在通知者手里于是它先阻塞在锁上等通知者释放后再继续这种模式被称为hurry up and wait。把通知放到解锁之后通常更好但需要保证共享状态已经更新完毕。正确性不受影响影响的只是调度效率。三、完整实现下面的实现把容量检查、关闭协议和异常安全都考虑进去了。元素用std::deque存放头尾操作都是常数时间push接受按值参数再移动入队调用方传左值或右值都能工作。// 适用于 C17#include condition_variable#include cstddef#include deque#include mutex#include optional#include stdexcept#include utilitytemplate typename Tclass BlockingQueue {public:explicit BlockingQueue(std::size_t capacity): capacity_(capacity) {if (capacity_ 0) {throw std::invalid_argument(BlockingQueue capacity must be 0);}}// 队列持有互斥量禁止拷贝移动也会破坏等待者与对象的关联一并禁止BlockingQueue(const BlockingQueue) delete;BlockingQueue operator(const BlockingQueue) delete;BlockingQueue(BlockingQueue) delete;BlockingQueue operator(BlockingQueue) delete;// 阻塞入队队列已关闭时返回 falsevalue 不被消费bool push(T value) {std::unique_lockstd::mutex lock(mutex_);not_full_.wait(lock, [this] {return closed_ || queue_.size() capacity_;});if (closed_) {return false;}queue_.push_back(std::move(value));lock.unlock();not_empty_.notify_one();return true;}// 非阻塞入队满或已关闭时立即返回 falsebool try_push(T value) {std::lock_guardstd::mutex lock(mutex_);if (closed_ || queue_.size() capacity_) {return false;}queue_.push_back(std::move(value));// 这里持锁通知换取实现简单效率敏感时可改为解锁后通知not_empty_.notify_one();return true;}// 阻塞出队关闭且排空后返回 nulloptstd::optionalT pop() {std::unique_lockstd::mutex lock(mutex_);not_empty_.wait(lock, [this] {return closed_ || !queue_.empty();});if (queue_.empty()) {return std::nullopt;}T value std::move(queue_.front());queue_.pop_front();lock.unlock();not_full_.notify_one();return value;}// 非阻塞出队空或已关闭且为空时返回 nulloptstd::optionalT try_pop() {std::lock_guardstd::mutex lock(mutex_);if (queue_.empty()) {return std::nullopt;}T value std::move(queue_.front());queue_.pop_front();not_full_.notify_one();return value;}// 关闭后不再接受新元素已入队的元素仍可被取走void close() {{std::lock_guardstd::mutex lock(mutex_);closed_ true;}not_empty_.notify_all();not_full_.notify_all();}bool closed() const {std::lock_guardstd::mutex lock(mutex_);return closed_;}std::size_t size() const {std::lock_guardstd::mutex lock(mutex_);return queue_.size();}std::size_t capacity() const noexcept {return capacity_;}private:const std::size_t capacity_;std::dequeT queue_;mutable std::mutex mutex_;std::condition_variable not_empty_;std::condition_variable not_full_;bool closed_ false;};几个设计点值得说明。push的参数按值传递调用方传右值时发生一次移动构造传左值时发生一次拷贝进入函数后再移动进容器整体只多一次移动换来接口的统一。如果追求极致可以重载const T与T两个版本但通常没有必要。close()先解锁再notify_all这是为了避免被唤醒的线程在锁上排队。两个条件变量都要通知等待不满的生产者需要知道队列已关闭等待不空的消费者需要知道该醒来看一眼是否排空。pop()在队列为空且已关闭时返回std::nullopt。注意判断顺序先看容器是否为空而不是先看closed_。关闭之后容器里可能还有元素这些元素必须能被取完否则数据就丢了。capacity()不加锁是安全的因为capacity_是const成员构造完成后不再改变多个线程只读没有竞争。四、生产者与消费者的配合示例下面演示一个生产者线程、两个消费者线程的完整流程。生产结束后调用close()消费者把剩余元素处理完后自然退出。// 适用于 C17#include chrono#include iostream#include thread#include vectorstruct Task {int id 0;int payload 0;};int main() {BlockingQueueTask queue(4);std::thread producer([queue] {for (int i 0; i 20; i) {Task t{i, i * i};if (!queue.push(t)) {break;}}queue.close(); // 生产者结束后关闭队列});std::vectorstd::thread consumers;for (int c 0; c 2; c) {consumers.emplace_back([queue] {while (auto task queue.pop()) {// 处理任务这里只做占位(void)task-payload;}});}producer.join();for (auto th : consumers) {th.join();}return 0;}关闭的责任必须明确。约定由最后一个生产者关闭或者由一个管理者线程在所有生产者join之后关闭。如果每个生产者都在结束时调用close()先结束的那个会让其他生产者无法继续入队任务被静默丢弃。消费者循环写成while (auto task queue.pop())条件里声明的是一个std::optionalTask在布尔上下文中判断是否有值task-payload访问内部对象。这种写法比先判has_value()再取值简洁也更不容易漏判。常见坑点❌ 用std::queue加锁但不加条件变量消费者用while (q.empty()) {}空转✅ 用条件变量等待避免浪费 CPU也避免在单核上把生产者饿死。❌cv.wait(lock);不带谓词醒来就认为有数据✅cv.wait(lock, []{ return !q.empty() || closed; });虚假唤醒和丢失唤醒都能挡住。❌ 在pop()中先判断closed_再判断队列是否为空关闭时直接返回导致队列里剩余元素丢失✅ 先取空队列队列为空且已关闭才返回结束标志。❌ 关闭时只通知not_empty_忘记通知等待不满的生产者✅ 两个条件变量都notify_all否则生产者永久阻塞。❌ 析构函数里直接销毁队列而不先关闭等待中的线程访问已销毁的互斥量✅ 先close()join所有线程再销毁队列或让队列对象活得比所有线程都久。❌ 把std::optionalT的判断写成if (!task)又去解引用或在try_pop返回空时仍然使用*task✅ 先确认有值再访问或用task-配合条件判断。❌ 认为notify_one一定唤醒消费者✅ 标准只保证唤醒至少一个等待者正确性由谓词循环兜底不能依赖唤醒对象。❌ 在持锁状态下做耗时处理例如把任务写文件再解锁✅ 只在锁内完成队列操作把业务处理放到锁外缩短临界区。总结关注点推荐做法错误做法容器选择std::deque头尾 O(1)用std::vector反复erase(begin())等待方式带谓词的wait裸wait或忙等条件变量数量不满、不空各一个共用一个且只notify_one关闭协议置标志、双通知、排空后返回空直接销毁或丢弃剩余元素返回值std::optionalT用特殊值表示结束和业务值冲突临界区只包住队列操作持锁做 I/O 或回调一个简单的消息队列并不复杂难点全在边界满与空、关闭与排空、唤醒与虚假唤醒。把这三对关系写清楚代码量不大但行为可预测。需要提醒的是本文的实现只适用于单进程内的线程间通信跨进程需要共享内存加进程间同步原语跨机器则要用真正的消息中间件语义和失败模型都完全不同。