C++线程安全队列实现：从互斥锁到无锁队列的并发编程实践

1. 项目概述：为什么我们需要线程安全队列？

在C++多线程编程的世界里，数据共享与同步是永恒的核心挑战。想象一下，你正在构建一个高性能的服务器，一个实时数据处理系统，或者一个需要并行计算的科学模拟程序。多个线程（比如数据采集线程、计算线程、日志写入线程）需要高效、有序地交换数据。如果让它们直接去读写一个公共的变量或容器，场面很快就会失控——数据竞争、条件竞争、死锁，各种难以调试的“幽灵”错误会接踵而至。这时，一个设计精良的线程安全队列（Thread-Safe Queue）就成为了协调多线程工作的“交通枢纽”和“缓冲区”。

简单来说，线程安全队列就是一个允许多个线程同时进行入队（push）和出队（pop）操作，而不会导致数据损坏或逻辑错误的队列数据结构。它封装了底层的同步机制，为上层应用提供了一个简洁、安全的通信接口。无论是实现生产者-消费者模型，还是作为任务调度器，它都是构建健壮并发系统的基石。本文将从零开始，深入剖析C++中实现线程安全队列的多种方案，从最基础的互斥锁+条件变量，到更高效的无锁（Lock-Free）队列，并结合实际场景，分享我在项目中踩过的坑和总结出的最佳实践。无论你是刚接触多线程的新手，还是希望优化现有并发架构的老手，这篇文章都将提供可直接复用的代码和深入骨髓的原理理解。

2. 核心设计思路与方案选型

实现一个线程安全队列，核心目标是在保证数据一致性的前提下，尽可能提升并发性能。不同的应用场景对性能和功能的需求不同，因此衍生出了几种主流的实现方案。选择哪种方案，取决于你的具体需求：是追求极致的吞吐量，还是需要简单的阻塞语义，亦或是需要支持超时等待等高级功能？

2.1 方案一：基于互斥锁(std::mutex)与条件变量(std::condition_variable)的阻塞队列

这是最经典、最直观，也是新手最容易上手的实现方式。其核心思想是：用一把大锁（互斥锁）保护整个队列数据结构，用条件变量来协调生产者与消费者的等待/通知机制。

为什么选择这个方案？

1. 实现简单，逻辑清晰：锁的语义直接，条件变量的使用模式固定（wait, notify_one, notify_all），代码易于理解和维护。
2. 功能完整：天然支持阻塞等待。当队列为空时，消费者线程可以挂起等待，直到有数据到来，这完美契合了生产者-消费者模型。
3. 标准库支持：完全使用C++11/14/17标准库组件，无需第三方依赖，可移植性强。

它的劣势也很明显：

1. 锁粒度大：任何操作（push/pop）都需要获取唯一的互斥锁，在高并发场景下，锁竞争会成为性能瓶颈。
2. 可能引入优先级反转：如果低优先级线程持有锁，高优先级线程可能被阻塞。
3. 不适用于所有场景：对于某些极低延迟或实时性要求极高的场景，阻塞和锁开销可能不可接受。

尽管如此，对于绝大多数需要清晰逻辑、稳定优先的应用（如后台任务调度、日志系统、普通的消息传递），这个方案依然是首选。它提供了一个可靠的基础，我们后续的许多优化都是在此基础上进行的。

2.2 方案二：细粒度锁或锁分离（Lock Stripping）队列

当单一互斥锁成为瓶颈时，一个自然的想法是：能不能减小锁的粒度？锁分离就是一种尝试。例如，可以为队列的头指针和尾指针分别配备独立的互斥锁。这样，一个线程在队尾插入（需要锁尾指针）时，另一个线程可以同时在队头弹出（需要锁头指针），只要它们操作的不是同一个节点，理论上就可以并发。

为什么考虑这个方案？

1. 提升潜在并发度：在理想情况下（队列非空且非满），入队和出队操作可以真正并行。
2. 思想具有启发性：它是从粗粒度锁向更高效并发数据结构演进的重要一步。

然而，它的实现复杂度和陷阱远超方案一：

1. 实现极其复杂：你需要精心设计节点结构，确保在同时修改头尾指针时，队列状态依然一致。处理空队列、单元素队列等边界条件时，锁的获取顺序可能引发死锁。
2. 收益未必显著：在队列较短或操作非常频繁时，两把锁带来的开销和复杂度可能抵消了并发带来的收益。现代操作系统和标准库对std::mutex的优化已经很好，简单的方案一在很多时候已经足够快。
3. C++标准库未提供：你需要自己从头实现整个链表和锁逻辑，容易出错。

因此，除非你经过 profiling 确认锁竞争确实是主要性能热点，并且有足够的信心和测试覆盖，否则不建议初学者或普通项目贸然使用此方案。它更像是一个高级的练习题，而不是通用的解决方案。

2.3 方案三：无锁（Lock-Free）队列

这是性能追求的终极方案之一。无锁队列通过原子操作（Atomic Operations）和内存序（Memory Order）来保证数据一致性，完全摒弃了互斥锁。线程间通过“比较并交换”（Compare-And-Swap, CAS）等原子指令来竞争修改指针，失败则重试。

为什么需要无锁队列？

1. 消除锁开销：没有锁，就没有了线程挂起、上下文切换的开销，对于核心数多的CPU和极高并发场景，性能提升可能非常显著。
2. 避免死锁和优先级反转：由于不使用锁，自然避免了这些由锁引发的问题。
3. 提供进度保证：在无锁算法中，至少有一个线程能在有限步骤内完成操作，系统整体不会因为某个线程挂起而停滞。

但是，它的代价是巨大的：

1. 实现极端复杂：正确的无锁算法是并发编程中最难的部分之一。你需要深入理解处理器内存模型（如x86的TSO，ARM的弱内存模型）、C++的std::memory_order（顺序一致性、获取-释放、松散顺序），并正确处理ABA问题。
2. 调试犹如噩梦：由于bug可能只在特定硬件、特定并发交错下出现，重现和调试极其困难。
3. 并非永远最快：在低竞争或简单场景下，无锁队列的开销可能比一个精心优化的有锁队列还要大，因为CAS操作本身也有开销，且失败重试会消耗CPU周期。

我的建议是：对于99%的应用，方案一（有锁队列）完全够用。只有在你构建像Disruptor那样的超高吞吐金融交易系统、核心游戏引擎，或像DPDK那样的网络数据平面时，才需要考虑自己实现或使用成熟的第三方无锁队列库（如boost::lockfree::queue或moodycamel::ConcurrentQueue）。

注意：盲目追求“无锁”是一种常见的性能优化误区。正确的优化流程永远是：1. 编写清晰正确的代码（通常用锁）。2. 进行性能剖析（Profiling），找到真正的热点。3. 针对热点进行优化。锁竞争往往是结果，而不是原因，可能是算法或架构设计需要调整。

3. 核心细节解析与实操要点

我们将聚焦于最实用、最可靠的方案一：基于std::mutex和std::condition_variable的阻塞队列，进行深度拆解。理解这个“标准答案”的每一个细节，是迈向高级并发编程的必经之路。

3.1 数据结构的选择：std::queue vs 自定义链表

队列的底层容器如何选择？std::queue默认使用std::deque作为底层容器，这带来了一个隐藏的陷阱。

问题在于std::queue::pop()的接口设计：它返回void，而通过front()获取队首元素。这意味着一个非线程安全的典型用法是：

if (!q.empty()) { auto value = q.front(); // 读取队首元素 q.pop(); // 移除队首元素 }

在多线程环境下，empty()、front()、pop()三个操作之间可能被其他线程打断，导致数据竞争或访问已释放内存。

因此，我们的线程安全队列必须提供一个“原子性”的操作：pop并返回队首元素。这引导我们放弃直接使用std::queue的pop()，而是选择以下两种方式之一：

1. 使用std::queue，但封装新的try_pop和wait_and_pop方法：在锁的保护下，先检查再获取再弹出。这是最常见的做法。
2. 使用自定义的单链表：每个节点动态分配，队头head_和队尾tail_（或dummy node）用指针管理。pop时，修改head_指针并返回原头节点的数据。这种方式在实现无锁队列时几乎是必须的，但在有锁队列中，用std::queue更简单。

为了教学清晰和向无锁队列过渡，我们这里采用带哑元节点（Dummy Node）的单链表实现。哑元节点是一个始终存在的空节点，它使得头尾指针的操作逻辑更统一，简化了空队列和单元素队列的边界处理。

3.2 条件变量的正确使用姿势

条件变量std::condition_variable是与互斥锁配合使用的同步原语，用于让线程等待某个条件成立。它的使用有几个关键陷阱：

1. 虚假唤醒（Spurious Wakeup）：即使没有线程调用notify_one或notify_all，等待的线程也可能被操作系统唤醒。这是POSIX标准和C++标准允许的行为。因此，条件变量的等待必须放在一个循环中，并反复检查等待条件。

std::unique_lock<std::mutex> lk(mutex_); // 错误：可能被虚假唤醒，此时data_queue可能仍是空的 // cv_.wait(lk, []{ return !data_queue.empty(); }); // 正确：使用lambda表达式作为谓词，wait内部会循环检查 cv_.wait(lk, [this]{ return !data_queue.empty(); });

wait的第二个参数（谓词）就是为了优雅地解决这个问题。在内部，如果谓词返回false，wait会释放锁并阻塞；被唤醒后，它会重新获取锁并再次检查谓词，只有谓词为true时才返回。

2. 通知丢失（Lost Wakeup）：如果在调用wait之前，另一个线程就调用了notify_one，那么这个通知可能会被“丢失”，导致等待线程永远阻塞。使用带谓词的wait并在持有锁的情况下修改条件和发出通知，是避免此问题的标准模式。

3. 条件变量的作用范围：通常，一个条件变量应与一个特定的条件（如“队列非空”）绑定。如果你需要等待多个条件（如“队列非空”和“队列未满”），可能需要多个条件变量，以避免“惊群效应”（所有等待线程被唤醒，但只有一个能继续工作）。

在我们的队列中，我们至少需要两个条件变量：一个给消费者，等待“队列非空”（cv_not_empty_）；另一个给生产者，如果队列有大小限制，则等待“队列未满”（cv_not_full_）。

4. 实操过程与核心环节实现

下面，我们将实现一个功能完整的线程安全队列模板类ThreadSafeQueue。它支持阻塞等待、超时等待、尝试弹出等功能。

4.1 类定义与数据结构

#include <memory> #include <mutex> #include <condition_variable> template<typename T> class ThreadSafeQueue { private: // 内部节点结构 struct Node { std::shared_ptr<T> data; // 使用shared_ptr存储数据，避免拷贝开销和异常安全问题 std::unique_ptr<Node> next; // 使用unique_ptr自动管理节点内存 Node() : data(nullptr) {} // 哑元节点构造函数 explicit Node(T value) : data(std::make_shared<T>(std::move(value))) {} // 数据节点构造函数 }; // 队列头尾指针。head_指向哑元节点，tail_指向最后一个节点。 std::unique_ptr<Node> head_; Node* tail_; mutable std::mutex head_mutex_; // 保护head_指针（理论上可与tail分离，此处为简化使用一把锁） mutable std::mutex tail_mutex_; // 保护tail_指针 std::condition_variable cv_not_empty_; // 队列非空条件变量 // 辅助函数：获取尾指针（需锁保护） Node* get_tail() const { std::lock_guard<std::mutex> tail_lock(tail_mutex_); return tail_; } // 辅助函数：在锁保护下弹出头节点（非线程安全接口，供内部调用） std::unique_ptr<Node> pop_head() { std::unique_ptr<Node> old_head = std::move(head_); head_ = std::move(old_head->next); return old_head; } // 等待队列非空，并返回锁（unique_lock） std::unique_lock<std::mutex> wait_for_data() { std::unique_lock<std::mutex> head_lock(head_mutex_); cv_not_empty_.wait(head_lock, [this]{ // 这里判断的是head_->next是否为空，因为head_是哑元节点 return head_.get() != get_tail(); }); return head_lock; // 返回锁，调用者持有锁，可以安全操作head_ } public: ThreadSafeQueue() : head_(std::make_unique<Node>()), tail_(head_.get()) {} // 初始化时head_和tail_都指向哑元节点 ThreadSafeQueue(const ThreadSafeQueue&) = delete; // 禁止拷贝 ThreadSafeQueue& operator=(const ThreadSafeQueue&) = delete; // 禁止赋值 // 尝试从队列前端弹出元素（非阻塞） std::shared_ptr<T> try_pop() { std::lock_guard<std::mutex> head_lock(head_mutex_); if (head_.get() == get_tail()) { // 队列为空 return std::shared_ptr<T>(); } return pop_head()->data; } // 等待并从队列前端弹出元素（阻塞） std::shared_ptr<T> wait_and_pop() { std::unique_lock<std::mutex> head_lock = wait_for_data(); // 等待并获得锁 return pop_head()->data; // 在锁保护下弹出 } // 入队操作 void push(T new_value) { // 在堆上构造新数据 std::shared_ptr<T> new_data(std::make_shared<T>(std::move(new_value))); // 构造新节点（此时还未链接入队列） std::unique_ptr<Node> p(std::make_unique<Node>()); { // 锁定尾指针，准备修改队列 std::lock_guard<std::mutex> tail_lock(tail_mutex_); tail_->data = new_data; // 将数据放入当前尾节点（原哑元或最后一个节点） Node* const new_tail = p.get(); tail_->next = std::move(p); // 将新哑元节点链接到队列末尾 tail_ = new_tail; // 更新尾指针指向新的哑元节点 } // 尾锁自动释放 cv_not_empty_.notify_one(); // 通知一个等待的消费者 } // 判断队列是否为空 bool empty() const { std::lock_guard<std::mutex> head_lock(head_mutex_); return head_.get() == get_tail(); } };

代码关键点解析：

1. 哑元节点模式：head_始终指向一个哑元节点。队列的第一个有效数据存储在head_->next节点中，或者当队列为空时，head_和tail_指向同一个哑元节点（其data为nullptr）。push操作会将数据放入当前tail_指向的节点（即旧的哑元节点），并在其后链接一个新的哑元节点，然后更新tail_。这样，empty()的判断就简化为head_.get() == get_tail()。
2. 数据存储使用std::shared_ptr<T>：这带来了两大好处。一是避免了在队列内部进行可能抛出异常的拷贝构造（数据在入队时由make_shared构造一次）。二是pop操作返回shared_ptr，使得数据的所有权可以安全、高效地转移给消费者，即使队列内部还在进行其他操作。
3. 节点内存管理使用std::unique_ptr<Node>：利用RAII机制自动管理节点内存，当节点被pop出队列后，unique_ptr会随着函数返回而销毁，自动释放节点内存，完全避免了内存泄漏。
4. 两把锁的简化使用：这里虽然声明了head_mutex_和tail_mutex_，但在push和try_pop中，为了简化演示，我们实际上通过锁的顺序获取（先head_mutex_再tail_mutex_，或反之）来避免死锁，并且empty()和wait_for_data()中也需要同时判断头尾。在更严格的实现中，get_tail()的调用也需要考虑锁的粒度。本例旨在展示思想，一个高度优化的两锁队列实现更为复杂。对于大多数情况，使用一把大锁保护整个链表结构（head_和tail_）是更简单安全的选择。
5. 条件变量的通知：push操作在修改队列后，释放尾锁，然后调用cv_not_empty_.notify_one()。注意，通知操作是在锁外进行的。这是一个重要的优化。如果在锁内通知，被唤醒的线程会立刻尝试获取锁，而此时通知线程可能还未释放锁，导致不必要的竞争和上下文切换。在锁外通知可以提高并发性能。

4.2 更完整的接口：支持超时等待和移动语义

一个工业级的队列还需要更多功能。以下是扩展接口示例：

// 在类定义中添加 public: // 带超时的等待弹出 std::shared_ptr<T> wait_and_pop_for(std::chrono::milliseconds timeout) { std::unique_lock<std::mutex> head_lock(head_mutex_); // 使用wait_for，并带谓词防止虚假唤醒 if (cv_not_empty_.wait_for(head_lock, timeout, [this]{ return head_.get() != get_tail(); })) { // 等待成功（条件在超时前满足） return pop_head()->data; } else { // 超时，返回空指针 return std::shared_ptr<T>(); } } // 尝试弹出到引用参数（避免动态内存分配，如果T类型拷贝成本低） bool try_pop(T& value) { std::lock_guard<std::mutex> head_lock(head_mutex_); if (head_.get() == get_tail()) { return false; } value = std::move(*pop_head()->data); // 移动数据 return true; } // 支持移动构造的push void push(T&& new_value) { std::shared_ptr<T> new_data(std::make_shared<T>(std::forward<T>(new_value))); // ... 后续链接逻辑与普通push相同 }

5. 常见问题与排查技巧实录

即使有了一个看似完美的线程安全队列，在实际使用中还是会遇到各种意想不到的问题。下面是我在项目中总结的几个典型“坑”和解决方法。

5.1 性能瓶颈分析与优化

问题场景：在一个高频交易模拟器中，使用了线程安全队列传递订单对象。性能测试发现，当生产者/消费者线程超过8个时，吞吐量不再增长，CPU使用率却很高。

排查与解决：

1. 使用性能分析工具（如perf, VTune）：发现大量的CPU周期消耗在std::mutex的锁竞争和std::condition_variable::wait/notify的系统调用上。
2. 优化锁竞争：
   • 批量操作：如果业务允许，生产者可以累积多条消息后一次性push，消费者也可以尝试一次pop多条（try_pop循环）。这显著减少了锁的获取/释放次数。
   • 使用更轻量的锁：在Linux下，pthread_spinlock_t（自旋锁）对于锁持有时间极短（纳秒/微秒级）的场景可能比std::mutex（互斥锁，可能引起上下文切换）更快。但要注意，自旋锁在单核CPU或锁持有时间长的情况下性能极差。C++11没有标准自旋锁，可使用std::atomic_flag实现或使用平台特定API。
   • 考虑无锁队列：这是终极方案。我们评估后引入了moodycamel::ConcurrentQueue（一个优秀的第三方无锁队列库），在超高并发（32+线程）下获得了近2倍的吞吐量提升。
3. 优化通知机制：
   • 避免不必要的通知：在push时，如果队列在push前就是非空的，说明已经有消费者在运行或即将被通知，此时notify_one可能不是必须的。但为了逻辑简单和正确性，通常保留通知。
   • 使用notify_all要谨慎：除非你确定需要唤醒所有等待线程，否则使用notify_one。notify_all会唤醒所有等待线程，它们会竞争锁，只有一个能成功，其他又会睡眠，造成“惊群”浪费。

5.2 队列积压与内存管理

问题场景：一个日志服务，生产者（多个业务线程）产生日志的速度远大于消费者（单个磁盘写入线程）的处理速度，导致队列中的std::shared_ptr日志对象堆积，内存持续增长直至OOM（内存溢出）。

排查与解决：

1. 实现有界队列（Bounded Queue）：为队列设置一个最大容量。当队列满时，push操作可以阻塞（wait在另一个条件变量cv_not_full_上），或者返回失败（try_push）。这迫使生产者降速，实现背压（Backpressure）。

   // 在类中添加容量限制和相关成员 size_t capacity_; std::condition_variable cv_not_full_; // push逻辑修改：在锁内检查 if (queue_size >= capacity_) { cv_not_full_.wait(...); } // pop逻辑修改：在成功pop后，调用 cv_not_full_.notify_one();

2. 使用std::unique_ptr代替std::shared_ptr：如果数据弹出后所有权完全转移，不需要共享，使用unique_ptr可以减少引用计数的开销。但接口需要调整，例如try_pop返回std::unique_ptr<T>。
3. 丢弃策略：对于日志这种可容忍丢失的场景，可以实现一个“丢弃最老”或“丢弃最新”的策略。当队列满时，新的push会覆盖最老的元素。这需要底层使用环形缓冲区（如std::vector）而非链表。

5.3 线程安全队列的销毁与线程退出

问题场景：程序退出时，消费者线程可能还在wait_and_pop上等待，而队列对象可能即将被销毁。直接销毁队列会导致在条件变量上等待的线程访问已释放内存，引发未定义行为（通常是崩溃）。

排查与解决：

1. 优雅关闭模式：在队列类中添加一个std::atomic<bool>标志位done_。

   std::atomic<bool> done_{false}; void shutdown() { done_.store(true); cv_not_empty_.notify_all(); // 唤醒所有等待线程 }

2. 修改wait_for_data或wait_and_pop：在等待条件中增加对done_的检查。

   cv_not_empty_.wait(head_lock, [this]{ return head_.get() != get_tail() || done_; }); // 在wait_and_pop中，被唤醒后检查： if (head_.get() != get_tail()) { return pop_head()->data; } else { return std::shared_ptr<T>(); // 返回空指针，表示关闭 }

3. 主线程协调：在程序退出前，主线程先调用queue.shutdown()，然后等待（join）所有消费者线程结束。消费者线程在收到空指针后，应退出其工作循环。

5.4 异常安全

问题场景：push操作中，构造std::shared_ptr<T>或new Node时可能抛出std::bad_alloc异常（内存不足）。如果异常发生在锁已获取之后，会导致锁无法释放，其他线程永久阻塞。

排查与解决：

1. RAII是救星：我们使用了std::lock_guard和std::unique_lock，它们会在析构时自动释放锁。即使push函数中发生异常，栈回滚也会调用这些锁守卫的析构函数，从而释放锁。这是使用RAII管理锁的核心优势之一。
2. 数据构造前置：注意我们的push实现：先在锁外构造好new_data（shared_ptr）和新节点p。锁只保护了将节点链接入队列和更新指针的极短操作。这进一步缩小了临界区，减少了异常发生在锁内的概率，也提升了并发性能。
3. 确保notify_one不被异常跳过：notify_one()本身是noexcept的，不会抛出异常。只要确保它在锁外调用，就不会有问题。

6. 进阶话题：无锁队列浅析与选型建议

虽然我们建议大多数项目使用有锁队列，但了解无锁队列的基本原理和选型标准是很有价值的。

6.1 无锁队列的核心：CAS操作

无锁算法的基石是原子比较交换操作。在C++中，这通过std::atomic类型的compare_exchange_strong或compare_exchange_weak成员函数实现。

std::atomic<Node*> head_; Node* old_head = head_.load(std::memory_order_relaxed); do { // 计算新值... new_node->next = old_head; } while (!head_.compare_exchange_weak(old_head, new_node, std::memory_order_release, std::memory_order_relaxed));

这段代码尝试将head_从old_head原子地更新为new_node。如果在此期间head_被其他线程修改（不再是old_head），则compare_exchange_weak失败，返回false，并将head_的当前值加载到old_head中，循环重试。这就是“无锁”中的“锁”的含义——通过硬件指令实现的乐观锁。

6.2 内存序：看不见的战场

std::memory_order指定了原子操作周围的内存可见性顺序。这是无锁编程中最复杂的部分。

• memory_order_seq_cst（顺序一致性）：最强约束，性能开销最大。所有线程看到的操作顺序一致。初学者和无把握时就用这个，它是默认值。
• memory_order_acquire/memory_order_release/memory_order_acq_rel：用于建立线程间的“同步-发生前”关系，性能更好。通常，加载用acquire，存储用release。
• memory_order_relaxed：最弱约束，只保证原子性，不保证顺序。用于计数器等场景。

错误的内存序会导致极难重现的数据竞争问题。强烈建议在彻底理解C++内存模型前，不要轻易使用relaxed或混合序。

6.3 第三方无锁队列库推荐

自己实现一个正确的无锁队列非常困难。在确实需要时，应优先考虑成熟的开源库。

1. boost::lockfree::queue/boost::lockfree::spsc_queue：

   • spsc_queue（单生产者单消费者）是性能最高的，实现相对简单。
   • 通用queue支持多生产者多消费者（MPMC）。
   • 优点：属于Boost库，质量高，文档相对齐全。
   • 缺点：接口有时不够灵活，某些场景下性能不是最优。

2. moodycamel::ConcurrentQueue：

   • 一个非常流行的高性能MPMC队列，作者为游戏行业开发。
   • 优点：性能极佳，提供了丰富的接口（阻塞、超时、批量操作等），使用起来比Boost的更顺手。
   • 缺点：非标准库，需要单独引入。

选型建议：

• SPSC场景：首选boost::lockfree::spsc_queue或简单的有锁环形缓冲区。
• MPMC场景，性能要求极高：首选moodycamel::ConcurrentQueue。
• 通用需求，希望减少依赖：使用本文实现的基于锁的队列，它简单、可靠、足够快。

7. 实战：集成到生产者-消费者模型

理论最终要落地。下面是一个简单的示例，展示如何使用我们的ThreadSafeQueue构建一个经典的生产者-消费者模型。

#include <iostream> #include <vector> #include <thread> #include <chrono> #include "thread_safe_queue.h" // 假设我们的队列实现在此头文件 void producer(ThreadSafeQueue<int>& queue, int id, int num_items) { for (int i = 0; i < num_items; ++i) { int value = id * 1000 + i; // 生产一个数据 queue.push(value); std::cout << "Producer " << id << " pushed: " << value << std::endl; std::this_thread::sleep_for(std::chrono::milliseconds(10)); // 模拟工作耗时 } std::cout << "Producer " << id << " finished." << std::endl; } void consumer(ThreadSafeQueue<int>& queue, int id) { while (true) { // 使用带超时的pop，方便优雅退出（实际应用中应有退出标志） auto data = queue.wait_and_pop_for(std::chrono::milliseconds(100)); if (data) { std::cout << "Consumer " << id << " popped: " << *data << std::endl; // 处理数据... std::this_thread::sleep_for(std::chrono::milliseconds(20)); // 模拟处理耗时 } else { // 超时，可以检查全局退出标志 // if (global_shutdown_flag) break; std::cout << "Consumer " << id << " timeout, checking shutdown..." << std::endl; } } } int main() { ThreadSafeQueue<int> queue; const int num_producers = 2; const int num_consumers = 3; const int items_per_producer = 5; std::vector<std::thread> producers; std::vector<std::thread> consumers; // 启动生产者线程 for (int i = 0; i < num_producers; ++i) { producers.emplace_back(producer, std::ref(queue), i, items_per_producer); } // 启动消费者线程 for (int i = 0; i < num_consumers; ++i) { consumers.emplace_back(consumer, std::ref(queue), i); } // 等待所有生产者结束 for (auto& t : producers) { t.join(); } // 在实际应用中，这里应设置退出标志，并等待队列清空或超时后，通知消费者退出。 // 本例为演示，消费者线程是死循环，主线程会一直等待。 // 可以按Ctrl+C终止程序。 std::cout << "All producers joined. Consumers are still running (Ctrl+C to exit)." << std::endl; // 简单演示：主线程等待一段时间，然后程序结束（消费者线程被强制终止，这不是好做法） std::this_thread::sleep_for(std::chrono::seconds(2)); // 更好的做法是实现优雅关闭，如第5.3节所述。 std::cout << "Main thread exiting." << std::endl; return 0; }

这个例子展示了队列如何作为线程间通信的桥梁。生产者线程独立产生数据，消费者线程独立处理数据，队列自动处理了它们之间的同步问题。在实际项目中，你还需要考虑线程池、优雅关闭、任务结果返回等更复杂的模式。

线程安全队列是多线程编程中的一个基础且强大的工具。从一把锁的简单保护，到精细的两锁分离，再到挑战智力极限的无锁算法，其演进过程本身就反映了并发编程的发展脉络。我的经验是，先从最经典、最可靠的模式入手，把它用熟、用透，理解其每一个细节和背后的权衡。当它真正成为你性能瓶颈时，你自然会有足够的动力和知识去探索更高级的解决方案。记住，在并发编程中，正确性永远比性能更重要。一个慢但正确的程序可以优化，一个快但充满竞态条件的程序，则是埋藏在系统中的定时炸弹。