简介：面向C++开发者的工业级无锁并发队列实现，基于C++11标准，专为多生产者、多消费者场景设计，提供无锁、线程安全的高吞吐队列。库采用单头文件实现，可整体嵌入项目，支持移动语义、批量操作与阻塞版本，并具备异常安全性，适合对延迟敏感的服务端、实时系统等低竞争环境。资源包共1828个文件，主体为1198个hpp头文件、339个h和92个cpp源文件，同时包含Visual Studio工程文件（sln/vcxproj等）及makefile等跨平台构建脚本，压缩包仅2.65MB。核心实现涵盖内存预分配、动态扩容机制，并与Boost、Intel TBB等无锁队列进行了对比，便于开发者评估不同方案；随附工程配置和辅助脚本可直接编译运行，帮助快速上手与二次开发。目前已有4240人学习下载，适合需要深度理解并发队列实现或在高并发项目中引入成熟无锁队列的C++工程师。 做过多线程流水线的人，大概都有过这样的经历：生产者和消费者线程明明各自干着轻量级活，数据量也不算大，但整个程序就是卡得莫名其妙。用性能分析器一看，锁竞争占了七八成开销。互斥锁在低竞争时确实便宜，一旦多个线程同时挤进来，内核态切换、缓存行颠簸、线程休眠唤醒这些成本就全出来了。这里要聊的 moodycamel::ConcurrentQueue，是一个用 C++11 实现的快速多生产者、多消费者（MPMC）无锁并发队列，单头文件引入即可使用。它的典型使用场景包括线程池任务分发、日志异步落盘、消息转发缓冲、事件总线这类需要跨线程传递数据的系统。如果你正在为锁竞争头疼，或者单纯想了解无锁队列在生产环境怎么落地，这篇文章值得读完。
1. 为什么需要一个无锁的 MPMC 队列：从锁竞争说起
1.1 锁不是慢在高竞争，而是慢在“被迫等待”
很多人对无锁队列的第一反应是“锁也能用，没必要搞复杂”。确实，单生产者单消费者场景下，一个带条件变量的环形缓冲区就足够高效，根本用不上无锁。但一旦变成多生产者多消费者，事情就变了。多个生产者同时调用push，锁必须保证互斥，这意味着同一时刻只有一个线程能进入临界区，其他线程全部自旋或休眠。
自旋本身不可怕，可怕的是自旋期间的缓存一致性开销。多个 CPU 核心同时读写同一把锁的缓存行，每次锁状态翻转都会触发缓存行在所有核心之间广播失效，这个成本远高于锁内临界区执行本身。CAS（Compare-And-Swap）指令也一样，无锁队列如果设计得不好，会让所有线程在同一地址上反复 CAS，效果和一把自旋锁没有本质区别。
ConcurrentQueue 的思路不是“消除所有原子操作”，而是尽量让不同生产者、不同消费者操作不同的内存地址，把竞争分散到各个独立的槽位和批次上。这样一来，即使队列整体吞吐很高，单个原子变量上的争抢强度也低得多。
1.2 谁适合用 ConcurrentQueue，谁不适合
用这个队列之前，先判断自己是不是目标用户。它适合的是线程数量较多、消息尺寸较小、对吞吐和延迟抖动有较高要求的场景。比如一个后台服务有 8 个网络线程接收请求，往队列里丢任务，后面 16 个工作线程取任务执行，这就是标准的 MPMC 模式。
如果你只有单生产者单消费者，直接用 moodycamel 家的另一个组件ReaderWriterQueue，它更轻、更快，专门为 SPSC 场景优化。如果业务上需要严格的消息顺序和背压控制，无锁队列反而不合适——无锁意味着没有阻塞机制，队列满了enqueue只会失败或返回 false，你得自己处理重试或丢弃策略。
1.3 无锁队列快在哪：核心是“分散竞争”和“减少等待”
ConcurrentQueue 在源码层面做了一系列针对高并发场景的优化。它把队列逻辑上拆分成多个子队列，每个生产者线程通过线程 ID 映射到不同的生产者槽位，写入时优先操作自己的槽位，避免了所有生产者挤在一把锁上。消费者端类似，通过ConsumerToken记录当前消费进度，批量取出元素，减少了原子操作的次数。
同时，它的内存序使用非常克制。C++11 提供了memory_order_relaxed、acquire、release、seq_cst等多个层级，seq_cst最安全但最慢，ConcurrentQueue 在设计时大量使用release/acquire配对来保证“写入完成后消费者能看到”，性能上比默认的全顺序一致性好不少。用无锁队列如果还用seq_cst满天飞，基本等于白做了无锁优化。
2. 核心接口与使用姿势：从入门到批量操作
2.1 最小的可用示例：一个生产者一个消费者
先看最直接的用法。ConcurrentQueue是头文件库，只需要把concurrentqueue.h放到项目里，然后包含进来就行，编译时打开 C++11 支持即可。
#include <atomic> #include <thread> #include <iostream> #include "concurrentqueue.h" int main() { moodycamel::ConcurrentQueue<int> q; std::thread producer([&] { for (int i = 0; i < 1000; ++i) { q.enqueue(i); } }); std::thread consumer([&] { int item; int count = 0; while (count < 1000) { if (q.try_dequeue(item)) { ++count; } } }); producer.join(); consumer.join(); std::cout << "done: " << count << std::endl; return 0; }这段代码有三点值得注意。第一，enqueue和try_dequeue都是非阻塞的，try_dequeue在队列为空时立刻返回 false，不会让线程睡着。第二，无锁队列没有“队列已满”的阻塞概念，内存不足时enqueue会尝试分配新块，极少数情况下可能抛出异常。第三，dequeue和try_dequeue的元素是拷贝语义，T 类型需要支持拷贝构造或移动构造，不能放引用。
测试时特意开多线程跑一下，生产 1000 万个 int，消费者用try_dequeue忙轮询，整体吞吐非常可观。忙轮询虽然看起来浪费 CPU，但在低延迟流水线场景里有时是必须的——线程宁可自旋也不要被唤醒，因为唤醒延迟通常有好几微秒甚至几十微秒。
2.2 多生产多消费者场景：Token 是关键
实际项目里很少有人只开一个生产者一个消费者，更多是多对多。直接用q.enqueue(x)和q.try_dequeue(item)也能工作，但性能不是最优。队列 API 为每个线程提供了 Token 机制，正确使用可以让线程在自己的本地缓冲上操作，减少跨核缓存同步。
moodycamel::ConcurrentQueue<int> q; // 每个生产者线程创建一个 ProducerToken 并复用它 moodycamel::ProducerToken ptok(q); q.enqueue(ptok, 42); // 每个消费者线程创建一个 ConsumerToken 并复用它 moodycamel::ConsumerToken ctok(q); int item; if (q.try_dequeue(ctok, item)) { // 处理这个 item }Token 的原理相当于给每个线程一个“专用车道”。生产者持有一个 token 后，写入时优先往自己上次使用过的子队列块里写，减少随机内存访问。消费者持有一个 token 后，能从上次读到的位置继续批量读取，而不必每次都从全局头部竞争。这个机制对单次 enqueue 来说可能只省几个纳秒，但在高并发下积少成多，提升非常明显。
创建 token 的时机也有讲究。最好在线程启动后创建一次并长期持有，不要每次入队都创建一个新 token，频繁构造 token 本身又变成另一种竞争。
2.3 批量操作：批量入队和批量出队
批量 API 往往是被忽视的性能利器。如果业务上经常一次处理一批消息，用try_dequeue_bulk一次性取出一批元素，能显著降低循环调用的函数开销和缓存缺失。
std::array<int, 64> items; size_t count = q.try_dequeue_bulk(ctok, items.data(), items.size()); for (size_t i = 0; i < count; ++i) { process(items[i]); }这个方法返回实际取出的元素数，可能小于请求的上限值。批量出队尤其适合带批处理的消费者：比如你要把数据攒一批再写入磁盘，一次拿到 64 条显然比循环 64 次单条出队高效。
批量入队相对少见，enqueue_bulk也是可用的：
std::vector<int> batch = {1, 2, 3, 4, 5}; q.enqueue_bulk(batch.data(), batch.size());2.4 Blocking 版本：需要等待语义时别自己写条件变量
concurrentqueue.h旁边通常还会带一个blockingconcurrentqueue.h，这是带阻塞语义的变体。如果你希望消费者在队列空时线程挂起而不是忙轮询，直接用这个版本最省心。
#include "blockingconcurrentqueue.h" moodycamel::BlockingConcurrentQueue<int> q; // 消费者线程阻塞等待 int item; q.wait_dequeue(item); // 超时版本 if (q.wait_dequeue_timed(item, std::chrono::milliseconds(100))) { // 取到了 } else { // 超时 }注意一点：BlockingConcurrentQueue 的入队路径依然是无锁的，但在队列空时消费者会通过条件变量挂起，这是它和纯无锁版本最大的区别。在低负载场景下，阻塞版本比忙轮询更省 CPU；在高吞吐场景下，忙轮询延迟更低。实际项目中我倾向于默认用 Blocking 版本，因为大多数业务系统 CPU 不是无限的，让消费者睡眠、由生产者唤醒，系统整体资源使用更健康。
3. 无锁队列的边界与陷阱：为什么踩坑的总是我
3.1 内存序不是万能的：对象生命周期必须自己保证
无锁队列最隐蔽的坑不是队列本身，而是元素对象的使用方式。ConcurrentQueue 内部通过无锁方式维护内存块和节点，它保证的是“节点指针的可见性”，但如果你把指针存进去，而指向的对象在消费者取出之前就被析构了，队列再快也救不了你。
典型错误是入队一个指向栈对象的指针：
moodycamel::ConcurrentQueue<Message*> q; void producer_bad() { Message msg; q.enqueue(&msg); } // msg 在这里析构了，可是消费者还没取走 void consumer() { Message* msg_ptr; if (q.try_dequeue(msg_ptr)) { msg_ptr->handle(); // 悬垂指针，崩溃 } }正确做法是入队堆对象的所有权，或者干脆入队值对象。无锁队列只保证数据的跨线程传递不产生数据竞争，不负责对象生命周期管理。这一点和std::queue配std::mutex的用法完全不同——加锁版本在锁内保证临界区的强顺序，而无锁版本只是把“有序发布”的责任交还给你了。
3.2 高并发下消费者可能饿死：没有公平性保证
多生产者多消费者场景下，无锁队列通常不保证公平性。也就是说，某些消费者线程可能长时间拿不到元素，而另一些消费者持续取走数据。虽然实际场景中这种情况很少发生，但如果你的业务依赖“每个消费者都能拿到差不多数量”的任务分片，就要小心了。
一个可行的缓解方式是：在消费者线程里给try_dequeue加超时重试策略，或者在业务层面对任务 ID 取模分片，先按模数分配给不同的子队列，再由各消费者消费固定子队列。这样能保证每个消费者的负载相对均衡。
3.3 固定容量和内存预分配
ConcurrentQueue不是完全固定的队列，它内部会按需创建新的块，但创建内存块本身是昂贵的。如果你明确知道队列中元素数量有上界，最好的做法是在构造时传一个合理的初始容量，减少运行期分配。
moodycamel::ConcurrentQueue<int> q(1024); // 预分配 1024 个元素的容量这里要澄清一个很容易误解的点：这个参数并不是说队列最多只能装 1024 个元素。它只是预分配初始块大小，队列仍然可以动态增长。更准确地说，这是为了让你提前分配足够多的初始内存，让前 1024 次 enqueue 不触发任何内存分配。
如果你拿 ConcurrentQueue 当固定容量有界队列用，比如实现一个“最多 10000 条，满了就丢”的缓冲，是不行的。它没有“满则拒绝入队”的语义，超额写入只会继续分配内存。要实现有界队列，你需要自己在外层用原子计数做限流。
3.4 与 ASan/TSan 配合：为什么 TSan 会告警
无锁队列在生产环境跑得好好的，但一开 ThreadSanitizer 就满屏告警，这种经历不少人遇到过。TSan 对无锁代码的检测依赖程序中的 happens-before 关系，moodycamel::ConcurrentQueue 内部用了正确的内存序，理论上不会触发虚假告警。但如果 TSan 仍然报出 data race，大概率是下面几种情况：
- 你入队的元素本身是一个包含非原子可变字段的结构体，生产者在入队后继续修改这个结构体；
- 你入队的对象在消费者读取前被另一个线程“抢先”修改了；
- 你把同一个对象同时入队到两个队列里，两个消费者各自读取。
解决办法是：入队后立刻交出所有权的对象，不允许再碰。这一点无论用不用无锁队列都是正确的多线程习惯，只是无锁场景下编译器不会帮你抓。
4. 工程化落地：编译、移植、性能调优
4.1 环境要求与编译配置
ConcurrentQueue 要求编译器支持 C++11。GCC 4.8+、Clang 3.4+、MSVC 2015+ 基本都行。在 CMake 里接入非常简单：
add_executable(main main.cpp) target_include_directories(main PRIVATE third_party/concurrentqueue) target_compile_features(main PRIVATE cxx_std_11)一个常见问题是：队列里的元素类型如果是一个非平凡析构的类（比如包含std::string），某些旧版本编译器上可能产生错误的告警。目前主流编译器配合 C++14/17 没有此类问题。如果你被困在老的 GCC 4.8 上，先升级编译器更现实。
4.2 性能对比：无锁不是银弹，但它在高竞争下很稳
我自己的测试环境是一台双路服务器，32 物理核，压测过三种实现：std::mutex+std::queue、自旋锁 + 数组队列、ConcurrentQueue。数据是 5000 万条整数消息，4 生产者 8 消费者。
低竞争（单生产者单消费者）时，ConcurrentQueue 和加锁版本的差距并不大，有时加锁版本延迟更低，因为无锁队列的批量分配策略在低吞吐下反而显得笨重。一旦线程数增多，竞争加剧，加锁版本的吞吐会迅速下降，线程切来切去光锁就能吃掉一半 CPU。ConcurrentQueue 的吞吐在高线程数下依然能保持平坦，这是它最大的价值。
如果你的业务是高吞吐低延迟并重，建议开启编译器自动向量化选项（-O2或/O2），并且避免把队列对象本身和热点数据放在同一个缓存行。多线程读写的队列对象如果和程序中的其他常变变量共享一个 64 字节缓存行，性能会莫名下降，这种伪共享问题用alignas(64)就可以避免。
4.3 一个实用的性能优化组合
我在实际项目中比较推荐这套组合：
- 每个线程只创建一个 ProducerToken / ConsumerToken，长期持有；
- 消息体用
std::unique_ptr或值类型，不要入队裸指针； - 消费者端用
try_dequeue_bulk批量取，攒够 N 条再统一处理； - 避免在入队和出队之间做任何日志打印或线性操作；
- 对队列容量和吞吐敏感的应用，用预分配初始容量，减少运行时内存分配。
这套组合做下来，大多数消息中间件的性能瓶颈已经不在队列本身了，而在于你后续的业务处理逻辑。无锁队列不是性能银弹，但它提供了一个非常低的开销基座，让你在排查性能问题时少一个怀疑对象。
5. 写在最后：无锁队列的取舍与实战体感
我个人用了两年多 ConcurrentQueue，最大的体会是：无锁队列适合做“流水线中间环节”，但不适合做“全局共享状态的协调器”。如果你需要的是线程间复杂的同步协作、多条件等待、任务优先级调度，那还是老老实实用条件变量和互斥锁，这些场景里无锁队列的弱项（无公平性、无阻塞、调试困难）会被无限放大。
反过来，只要你的业务模型是清晰的 生产-消费 模式，任务边界明确，消息生命周期可控，ConcurrentQueue 能在极小的代码量下带来非常可观的吞吐提升。尤其是当你发现程序里锁竞争占了 CPU 的 30% 以上的时候，换掉它，你往往能直观地感受到整体响应变得平滑。
最后再分享一个调试小技巧：无锁队列的 bug 非常难复现，所有问题都呈概率性出现，所以在开发阶段我会特意用 TSan 跑完整测试，再在低配多核机器上做高并发压测。如果队列上下游的对象生命周期和边界条件在压力下能稳定跑上几个小时，基本可以踏实交给线上。
本文还有配套的精品资源，点击获取
