★ FEATURED ARTICLE 简介moodycamel::ConcurrentQueue 是一个基于 C11 的工业级无锁并发队列实现专为多生产者、多消费者MPMC场景设计目标是替代传统互斥锁队列提升线程吞吐与响应速度。它采用单头文件形式可直接嵌入项目使用通过模板化支持任意元素类型并自动管理内存尽可能移动而非复制元素同时提供批量操作、低开销阻塞版本与异常安全保证。整个资源包共包含1828个文件以1198个hpp和339个h头文件为主另有92个cpp源文件及sln、vcxproj等工程配置makefile、asm等跨平台构建文件文件类型涵盖了源码、工程、构建脚本与说明文档压缩包仅2.65MB。目前已有4242人学习或浏览体积小巧却信息密度高。通过阅读源码和示例读者可以掌握无锁队列的原子操作、内存序、内存管理及批量入队出队的优化思路也可直接复用该头文件构建高吞吐并发系统若只需要单生产者、单消费者场景资源也提供了相应轻量实现。整体适合有一定C并发编程基础、想研究工业级无锁数据结构或优化队列性能的中级及以上开发者。1. 无锁MPMC队列多线程吞吐瓶颈下concurrentqueue值得先试当多线程程序里锁等待成为主要开销时把std::queue加互斥锁换成 concurrentqueue 这种无锁并发队列是最容易落地的改动之一。concurrentqueue 是一个基于 C11 标准实现的多生产者多消费者MPMC无锁并发队列它在同一时刻允许多个线程写入、多个线程读取却不依赖互斥锁而是用原子操作、内存序和分块内存保证正确性。它适合任务分发、日志聚合、消息管道这类数据小而密的场景能明显降低线程切换和锁竞争带来的延迟。如果你正在做中间件、网关或任务调度先把它用起来再逐步理解内部机制收益会很明显。2. 它为什么快无锁、分块与缓存友好的代价和边界2.1 三种并发队列的选型视角锁等待、自旋与无锁传统做法是std::queue加一把std::mutex。临界区很短时锁竞争不严重但一旦生产者和消费者线程数量都超过 2锁的等待和唤醒就会成为常态。线程被挂起再唤醒涉及内核调度单次可能花费微秒级时间而队列操作本身可能只需要几十纳秒。于是程序的时间大量耗在调度器上业务逻辑反而没跑多少。第二种做法是自旋锁队列用原子变量做 CAS 自旋拿不到锁就原地转圈。自旋的好处是不会挂起线程避免了内核切换但坏处是空转时 CPU 占用高多个线程同时抢一个自旋锁还会引发缓存行乒乓。临界区极短、线程数不超过 CPU 核数的场景下自旋锁有效一旦线程被抢占其他线程就得陪着等。concurrentqueue 走的是第三条路不锁队列本身而是把争用点拆散。生产者之间、消费者之间、生产者与消费者之间的互斥都落在独立的原子变量上并且通过分块机制让线程不是每写一个元素都要打一次全局同步而是先竞争到一个块然后在块内连续写入。下面是三种方案的对比方案同步方式空队列行为适用场景主要代价std::queue mutex互斥锁线程挂起或自旋低频大任务唤醒开销、锁竞争自旋锁队列原子 CAS 自旋忙等临界区极短、线程少CPU 空转concurrentqueue无锁原子 分块忙读但可批量读取高频小消息、MPMC内存占用、实现复杂选型时先看任务粒度如果每个任务都要做几十毫秒的 IO锁等待占比很小换无锁队列收益不大。反过来如果消息体很小、频率很高锁的固定开销就变成了瓶颈这时候 concurrentqueue 的价值最大。2.2 无锁的底层支撑CAS、内存序与伪共享无锁的核心是 CASCompare-And-Swap指令一个线程把某个内存位置的旧值读出来判断它仍然等于预期值时才写入新值整个过程是一条原子指令。concurrentqueue 里的生产者申请块、消费者推进读位置靠的都是这类原子操作。它们不需要锁但依赖 CPU 提供的缓存一致性协议。C11 的std::atomic给出了内存序参数。理解 concurrentqueue 的正确性不需要精通全部六种内存序但要抓住两点写入队列数据的线程用release发布读取数据的线程用acquire获取。这样能保证数据先写入内存再让索引被别人看到的顺序不会被编译器或 CPU 重排。如果用错成relaxed高并发下可能出现消费者看到索引更新、却没看到数据本身的情况这是无锁队列最容易翻车的地方。伪共享是另一个隐藏凶手。两个线程各自操作互不相关的原子变量如果这两个变量落在同一条 64 字节缓存行里其中一个线程的写入会导致另一个线程的缓存行失效于是出现明明没有共享数据却互相拖慢的现象。concurrentqueue 在设计上把高频写变量和高频读变量分隔开工程里如果你自己封装多队列实例也需要留意同样的陷阱。还有一点要纠正无锁不等于无等待。concurrentqueue 的消费者在队列为空时如果代码写成死循环try_dequeueCPU 会一直转。它只是把谁该等、等到什么时候交给调用方决定而不是在库内部挂起线程。这个特性后面会专门讲。2.3 块与槽位MPMC 队列如何摊薄原子操作concurrentqueue 没有把队列做成简单的环形缓冲而是用块作为基本单位。一个块包含一串槽位生产者拿到一个块的所有权后可以连续往槽位里写多个元素写完后更新块的发布计数。这样一来多个生产者竞争的粒度从每个元素变成了每个块原子操作次数直接除以块内元素数量。消费者侧是另一个方向。所有生产者写入的数据在逻辑上形成一个全局序列消费者按顺序跨块读取。每个消费者通过原子变量获取当前可读的块和槽位读走数据后更新自己的消费位置。多消费者场景下队列不需要把任务分配给某个消费者而是让消费者自己去争抢读位置这天然实现了负载均衡。分块设计还有一层好处内存增长按块分配扩容时不需要像std::vector那样搬移已有元素。块本身通过内部的内存管理器复用生产消费节奏稳定时堆分配次数很少性能曲线更平缓。理解这一点再看后面讲容量设置时会更容易。3. 接入与第一个 MPMC 程序最小示例、阻塞版与工程集成3.1 工程集成头文件选择、编译选项与目录组织这个库的接入方式很轻。官方发布同时提供源码版和单头文件版源码版适合想调试内部实现的场景单头文件版把concurrentqueue.h复制进include目录即可我用的一直是单头文件版省去链接步骤。阻塞版队列在另一个头文件里叫blockingconcurrentqueue.h它依赖concurrentqueue.h两个文件一起放进工程就行。编译上C11 是硬性要求。由于内部大量使用std::atomic、std::memory_order和可变参数模板编译器需要支持 C11 及以上标准。在 CMake 里设置set(CMAKE_CXX_STANDARD 11) set(CMAKE_CXX_STANDARD_REQUIRED ON)如果你还在用旧编译器至少需要支持 N2660 原子库的版本。另外建议开启优化编译无锁队列在-O2和-O0下的表现差别很大-O0下原子操作可能被编译器生成额外的屏障指令测出来的数字会误导你对库的判断。头文件引入方式#include concurrentqueue.h #include blockingconcurrentqueue.h using namespace moodycamel; // 库的命名空间注意命名空间不是std代码里写ConcurrentQueue之前要先using或者加前缀。这个细节初学者很容易忽略编译报错时第一反应往往是头文件没包含对实际是命名空间没引入。3.2 最小 MPMC 示例双生产者、双消费者读写下面这段代码演示最核心的用法两个生产者线程各入队 100 万个任务两个消费者线程同时读取最后统计消费总数。它不涉及阻塞等待完全用try_dequeue的返回值判断是否读到数据。#include atomic #include iostream #include thread #include vector #include concurrentqueue.h using namespace moodycamel; struct Job { int id; int payload; }; int main() { // 无锁 MPMC 队列默认容量按需增长 ConcurrentQueueJob q; constexpr int kProducers 2; constexpr int kConsumers 2; constexpr int kTasksPerProducer 1000000; // 记录已完成写入的生产者数量用于通知消费者退出 std::atomicint producers_done{0}; std::atomiclong long total{0}; std::vectorstd::thread producers; for (int t 0; t kProducers; t) { producers.push_back(std::thread([q, producers_done, t]() { for (int i 0; i kTasksPerProducer; i) { // enqueue 在队列容量不足时自动扩容通常不会失败 q.enqueue(Job{ t * kTasksPerProducer i, i % 233 }); } // release 保证之前入队的写入对其他线程可见 producers_done.fetch_add(1, std::memory_order_release); })); } std::vectorstd::thread consumers; for (int c 0; c kConsumers; c) { consumers.push_back(std::thread([q, producers_done, total]() { Job job; long long local 0; // 生产者全部结束前先通过 try_dequeue 忙读 while (producers_done.load(std::memory_order_acquire) kProducers) { if (q.try_dequeue(job)) { local; // 读到一个任务 } else { std::this_thread::yield(); // 队列暂时为空让出 CPU } } // 生产者结束后再清空队列里的残余任务 while (q.try_dequeue(job)) { local; } // 用 relaxed 累加即可统计准确性要求不高 total.fetch_add(local, std::memory_order_relaxed); })); } for (auto th : producers) th.join(); for (auto th : consumers) th.join(); std::cout consumed total.load() jobs std::endl; return 0; }代码里值得注意的有三处。第一enqueue和try_enqueue的区别前者是一定能入队的语义无界队列在内存充足时总会成功后者返回 bool适合在资源受限环境里做降级处理。第二消费者用try_dequeue的返回值判断有没有读到数据而不是先查size()再决定是否读取后者在并发下不可靠。第三producers_done用 release/acquire 配对保证消费者看到计数更新时队列里的数据一定已经完整写入。如果你改成memory_order_relaxed在这个示例里多半也能跑通但在更复杂的释放场景里可能出问题不建议省。运行结果应该是两百万个任务全部被消费。你可以把kProducers、kConsumers改成 4 或 8观察吞吐变化。如果发现消费者明显滞后通常不是队列的问题而是消费者线程里的处理逻辑太重或者任务分配不均匀。3.3 阻塞版 BlockingReaderWriterQueue消费者不想忙等时上一节的示例里队列为空时消费者只能yield让出 CPU本质还是轮流试探存在一定的空转浪费。如果你的业务要求消费者线程一旦空闲就彻底休眠等数据来了再被唤醒就用阻塞版队列。BlockingReaderWriterQueue提供wait_dequeue方法队列为空时调用线程挂起等待生产者入队后内部通过条件变量唤醒等待中的消费者。代价是每次唤醒都有系统调用和线程切换开销吞吐通常低于忙等版本但 CPU 占用低得多尤其适合消费者线程数量多、任务稀疏的场景。#include thread #include blockingconcurrentqueue.h using namespace moodycamel; struct Task { int seq; }; int main() { // 阻塞版队列生产者写入消费者用 wait_dequeue 等待 BlockingReaderWriterQueueTask bq; std::thread consumer([bq]() { while (true) { Task t; bq.wait_dequeue(t); // 队列为空时挂起有数据时被唤醒 if (t.seq -1) break; // 哨兵任务用于通知消费者退出 // 在这里处理实际业务 } }); // 生产 100 个正常任务 for (int i 0; i 100; i) { bq.enqueue(Task{i}); } // 生产哨兵任务消费者读到后退出 bq.enqueue(Task{-1}); consumer.join(); return 0; }这个示例里用到了一种常见退出模式用特殊值的任务作为哨兵。消费者只有在读到哨兵时才退出循环这样能保证前面的 100 个任务都被顺序处理完。如果你的任务结构里没有地方放哨兵值可以单独用一个原子 bool 做停止标志消费者在wait_dequeue和标志位之间二选一但要注意wait_dequeue可能长时间阻塞停止标志需要配合超时等待或额外唤醒。选型建议日志收集、事件驱动这类大部分时间没有数据的系统用阻塞版高吞吐数据管道、消息转发这类数据几乎始终在流动的系统用无界版加忙读。两者之间没有绝对优劣按任务密度决定。4. 参数设置容量、批量出入队与 token 的适用边界4.1 初始容量与动态增长小容量不是缺陷ConcurrentQueue的构造函数可以传初始容量例如ConcurrentQueueJob q(1024)。但很多人误以为这个值像std::vector::reserve那样决定性能上限其实并发队列的容量是按块增长的初始值只影响最早几个块的数量。容量不足时队列会分配新块块被消费完后进入内部空闲列表后续可复用不会立刻释放回系统。所以小容量不是缺陷反而有好处线程少、消息不多时初始化开销小内存占用低。真正需要关注的是任务峰值时的增长行为。如果生产速度短时间暴涨而队列里积压的数据又很大连续分配新块可能带来延迟抖动。我的做法是预估峰值积压量按峰值设置初始容量减少运行期分配次数。还有一点容易忽略队列里的空闲块内存不会无限膨胀。只要消费速度跟得上块会进入循环复用但如果生产端偶发爆发、之后长时间空闲队列会保留一批已分配的块。这个高水位内存占用是正常的不属于泄漏排查内存问题时别误判。4.2 批量出入队吞吐翻倍的最直接手段批量操作是 concurrentqueue 最容易见效的参数。一次enqueue调用只需要一次原子同步如果你能一次提交一批元素让库在块内连续处理原子操作次数不变但吞吐量线性上升。出队侧同理一次try_dequeue_bulk取出一批元素消费端可以在本地循环处理减少对全局读位置的争用。constexpr int kBatchSize 256; Job batch[kBatchSize]; // 生产者批量入队 for (int i 0; i kBatchSize; i) { batch[i] Job{i, i * 3}; } size_t enqueued q.try_enqueue_bulk(batch, kBatchSize); // 消费者批量出队 Job out[kBatchSize]; size_t got q.try_dequeue_bulk(out, kBatchSize); for (size_t i 0; i got; i) { process(out[i]); // 处理同一批任务 }try_enqueue_bulk和try_dequeue_bulk都返回实际处理的数量可能小于请求的批量大小调用方必须用返回值做循环条件不能假设got kBatchSize。批次大小的选择建议对齐缓存行常见值 256 或 512。太大时数组占用栈空间太小则发挥不出批量优势。如果你的元素本身超过 64 字节批次可以适当调小避免栈溢出和缓存行浪费。在日志聚合这类场景生产者往往自己就有攒批逻辑攒够 256 条再入队消费者读到批量任务后直接写文件或网络缓冲。这种两端都批量的设计配合无锁队列经常能比逐条入队快一倍以上。代价是增加了约一个批次的延迟实时性要求极高的系统要自己权衡。4.3 token 参数索引滞留在线程本地的收益与限制token 是 concurrentqueue 提供的一个进阶参数理解它需要回到伪共享这个概念。默认情况下线程每次入队都要读取全局生产者索引这个索引被所有线程共享。即使不产生逻辑竞争缓存行同步也会拖慢速度。token 的作用是让每个线程缓存部分索引信息减少对全局变量的访问频率。ConcurrentQueueJob q; // 每个线程创建自己的 token ConcurrentQueueJob::ProducerToken ptok(q); ConcurrentQueueJob::ConsumerToken ctok(q); // 生产者线程内带 token 入队 q.enqueue(ptok, Job{1, 2}); // 消费者线程内带 token 出队 Job j; if (q.try_dequeue(ctok, j)) { // 处理 j }使用 token 的规则很简单每个线程持有一个属于自己的 token不要在多个线程间共享同一个 token。token 必须在队列构造之后创建并绑定到队列实例。带 token 的版本和不带 token 的版本可以同时工作但同一线程内最好保持一致避免索引缓存错乱。值不值得用取决于你的线程模型。如果每个线程持续不断地读写同一个队列token 能明显降低原子变量访问频率如果线程偶尔才操作一次队列token 的收益可以忽略反而多一份代码复杂度。我一般在至少 4 个生产者线程、每个线程每秒操作超过 10 万次的场景里才引入 token低并发下测不出差别纯粹增加心智负担。5. concurrentqueue 避坑指南5 个真实场景下的踩坑记录5.1 析构丢数据退出顺序没编排好现象程序退出前队列里还有未处理的任务日志显示数据丢失。排查时发现队列对象先于消费者线程析构消费者还没读完内存块已经被释放。原因并发队列的析构不会等待消费者把剩余数据取完。无界版更不会在析构时优雅排空它只是释放内部资源。解决退出顺序必须是先停生产再消费排空最后析构队列。具体做法是先置一个停止标志生产者停止入队消费者循环读到队列为空后退出join所有消费者线程最后才让队列对象离开作用域。如果你用的是阻塞版注意在置停止标志后再入队一个哨兵任务把可能阻塞的消费者唤醒。5.2 忙等烧 CPUtry_dequeue 空转的代价现象消费者线程看起来在忙但业务处理量很低top显示线程 CPU 占用接近 100%。原因try_dequeue在队列为空时立即返回 false如果代码写成while(true) { if(q.try_dequeue(j)) ... }无论有没有数据循环都不会停止CPU 空转。解决低频场景直接换BlockingReaderWriterQueue用wait_dequeue挂起。高频场景保留忙读但要加入退让逻辑连续 N 次读不到数据时std::this_thread::yield()或sleep_for(1ms)。N 一般取 1000 左右既能保证数据到达时快速响应又避免空转。5.3 用 size() 判空不靠谱计数只是估算值现象消费者用q.size() 0判断队列已空退出后却发现还有任务没处理或者size()返回非零但try_dequeue已经拿不到数据。原因并发环境下精确维护大小需要额外同步concurrentqueue 出于性能考虑size()返回的是一个估算值可能滞后于实际入队出队操作。它适合监控积压趋势不适合做精确判空。解决一切以try_dequeue的返回值为准。需要统计还剩多少任务时自己在外部维护一个原子计数器入队加一、出队减一比依赖size()可靠得多。5.4 大对象进队列拷贝开销让收益变负现象队列元素换成体积很大的结构体比如带std::string和std::vector的复合类型后压测吞吐反而比std::queue mutex慢。原因concurrentqueue 默认按值存储入队时拷贝元素出队时移动元素。无锁省下的同步时间被大对象的内存拷贝开销抵消了。解决队列元素尽量放轻量数据比如把任务索引、指针放进队列实际数据存在外部共享结构里或者用std::unique_ptr包一层入队移动指针而不是深拷贝。如果必须放可拷贝大对象至少确认它实现了高效的移动构造函数-O2下移动比拷贝快得多。5.5 多队列互相拖累伪共享在实例之间也存在现象两个队列分别跑都很快放在同一个类里同时使用两者都变慢CPU 缓存命中率下降。原因类里两个队列对象紧挨着定义内部高频访问的原子变量可能落在同一条缓存行上互相影响导致缓存行频繁失效。解决定义队列成员时按缓存行对齐alignas(64)修饰队列成员或把它们放到独立堆分配的对象中。这个坑对单队列不成立但对一个线程管理多个队列的场景很常见。6. 压测与退出机制把吞吐量、延迟和收尾设计都量化6.1 压测框架用 ops/s 和延迟分位数说话判断并发队列值不值得换不能靠感觉要压测。我的压测方式固定两步先测吞吐再测延迟分布。吞吐用总处理量除以耗时延迟只测单条消息从入队到出队的间隔。#include chrono #include atomic auto begin std::chrono::steady_clock::now(); // 运行固定时长记录 total 完成量 // ... auto end std::chrono::steady_clock::now(); double sec std::chrono::durationdouble(end - begin).count(); std::cout (total.load() / sec) ops/s std::endl;对比对象用std::queue std::mutex同样的生产者消费者数量、同样的消息体。如果无锁版本跑不出显著优势说明瓶颈不在锁竞争可能在业务处理或 IO 上这时候换队列就是白费功夫。延迟则统计 p50、p99 和 max 三个值重点关注 p99因为锁竞争最伤的是尾部延迟。6.2 退出机制哨兵值与停止标志的配合多生产者多消费者模型的退出往往比启动更麻烦。单独用停止标志不够因为消费者可能正阻塞在wait_dequeue上单独用哨兵任务也不够因为多个消费者要区分谁收到哨兵。常见做法是每个消费者一个专属哨兵或者用停止标志 队列排空两步走先置标志生产者停止入队再向阻塞版队列投 N 个哨兵任务N 等于消费者线程数确保每个消费者都能被唤醒并退出。6.3 什么时候放弃无锁队列无锁队列不是银弹。任务粒度大、消息体复杂、线程数很少的场景锁的开销不占主导无锁的复杂度反而增加维护成本。如果业务需要优先级、严格公平或者任务可持久化concurrentqueue 这类通用队列实现不了不如在更高层做设计。这些年我养成的习惯是新项目先写一版阻塞队列原型压测确认瓶颈确实在锁竞争再换无锁队列做对比两版数据都留着方便后续性能回归。希望帮到你。本文还有配套的精品资源点击获取
阅读完成 · 觉得有帮助？
