做多线程开发，队列几乎是绕不开的组件。无论是生产者消费者模型、线程池任务调度、还是日志异步落盘，背后都要有一个安全可靠的队列来搬运数据。但很多人在自己写队列时，总会遇到灵异事件：单测全绿，一上多线程就偶发崩溃、卡死、数据丢失，把 bug 归结于玄学。其实这真不是玄学，而是没有理解“线程安全队列”和“普通队列 + 加锁”之间的本质区别。
这篇文章想把我这些年实现和使用线程安全队列的实战经验做一个完整梳理。我会从一个最普通的队列在并发场景下是怎么崩溃的讲起，逐步带你写出基于锁的可靠实现，再深入到无锁队列的内存序和 ABA 陷阱，最后给出实际工程中的选型建议和排错方法论。内容偏 C++，但思路对所有语言通用。适合已经会写基本多线程代码、但想真正搞明白“为什么我的队列不安全”和“怎么设计一个靠谱的队列”的开发者。
1. 为什么普通队列到了多线程环境就失控：竞态条件的根源
先说个最常见的现象：你写了一个漂亮的std::queue封装，提供push和pop两个接口，看起来天衣无缝。然后在两个线程里一个生产一个消费，跑起来居然时不时崩溃，或者 pop 拿到的数据和 push 进去的不一样。排错半天，最后发现是队列本身的问题。
要理解这个问题的本质，得先搞清楚一件事：你写的每一行高级语言代码，编译之后都是多条机器指令，而线程调度的切换，是可以在任意一条机器指令之间发生的。别把q.push(x)当成一个原子操作——它不是。
1.1 一个最简单的计数器问题，如何演变成队列灾难
我拿一个最简单的push流程举例。假设队列内部维护着尾指针、大小计数器和存储节点：
void push(int val) { tail->next = new Node(val); tail = tail->next; count++; }这段代码看起来很干净，但展开成机器指令后，大致会变成这么几个步骤：
- 读取
tail指针的值 - 在堆上分配一个节点
- 将新节点的地址写入
tail->next - 用新节点地址更新
tail->next对应的内存 - 更新全局的
tail指针本身 count加 1
如果两个线程同时调用push，调度器完全可能在某个线程执行完第 2 步、还没来得及执行第 3 步时，把 CPU 切换给另一个线程。第二个线程进来也会执行同样的读取tail、分配节点。等到第一个线程恢复执行，它更新的是旧tail，第二个线程也在旧tail上挂节点。结果就是：一个节点被覆盖丢失，链表指针错乱，或者计数对不上。
你可能觉得这种交错概率很低，但多线程程序跑的规模上来之后，有一点概率就足以让生产环境天天出事。这也是为什么"看起来没问题"的代码，在并发压力下总是崩得毫无预兆。
1.2 竞态窗口：指令交错才是问题的元凶
我们把上面的分析抽象成一个概念——竞态窗口。所谓竞态窗口，指的是从"读取共享数据"到"基于读取结果完成修改"这整段区间。这段区间内，只要有第二个线程也碰同一个共享变量，就会出问题。
在队列里，共享变量远不止一个，至少包括头指针、尾指针、节点计数、节点内部的 next 指针。单变量竞争已经很麻烦，多变量之间的交互竞争才真正致命。比如push修改尾指针的同时，pop正在读取头指针，某个中间状态下，两个指针可能指向同一个节点，于是消费线程把刚生产的节点给删了，生产线程还在操作它。这种内存破坏通常是完全随机的，连核心转储分析起来都极其恶心。
1.3 数据竞争与未定义行为：比你想的更严重
C++ 标准里有一条很硬核的规定：只要两个线程同时对同一个对象做读写操作，且至少有一个是写操作，并且没有同步机制，这就是数据竞争，而数据竞争属于未定义行为。未定义行为的含义是：编译器想怎么生成代码都行，程序想做任何事也都不算违规。
也就是说，竞态窗口下的崩溃不是"偶尔出错"，而是标准层面已经宣判你的代码不可信了。编译器在做优化时，完全可能把读写顺序重排、把共享变量缓存到寄存器里反复使用，最终产生的行为比你能想象的 bug 还要离谱。理解了这一点，后面的方案就都有依据了：
- 要么用锁把竞态窗口关掉
- 要么用原子操作和内存序规则，让竞态窗口变成无锁状态下的合法并发
2. 锁保护队列：最稳妥的基线实现与条件变量的正确用法
如果你想把队列做成线程安全的，第一个念头肯定是在每个接口前面加一个互斥锁，这方向没问题，但很多初学者在这儿会犯一个典型错误：只保护 push 和 pop 的临界区，却忽略了"消费者没数据时怎么办"这个问题。真实的消费者不可能永远自旋等待，必须有一种机制让它安全地睡过去，等有数据再被叫醒。这就是条件变量的工作。
2.1 一个完整的锁保护队列实现，拿来就能用
下面这段代码，是我在项目里常用的一个模板，兼顾了安全性、性能和可读性。为了让你看得更清楚，我用了 C++17 标准的写法，std::mutex负责互斥，std::condition_variable负责同步通知。
#include <queue> #include <mutex> #include <condition_variable> #include <optional> template <typename T> class ThreadSafeQueue { public: void push(const T& value) { { std::lock_guard<std::mutex> lock(mutex_); queue_.push(value); } not_empty_.notify_one(); } // 阻塞式弹出，没有数据就休眠等待 T pop() { std::unique_lock<std::mutex> lock(mutex_); not_empty_.wait(lock, [this] { return !queue_.empty(); }); T value = std::move(queue_.front()); queue_.pop(); return value; } // 非阻塞尝试弹出，没有数据立即返回 nullopt std::optional<T> try_pop() { std::lock_guard<std::mutex> lock(mutex_); if (queue_.empty()) { return std::nullopt; } T value = std::move(queue_.front()); queue_.pop(); return value; } size_t size() const { std::lock_guard<std::mutex> lock(mutex_); return queue_.size(); } bool empty() const { std::lock_guard<std::mutex> lock(mutex_); return queue_.empty(); } private: std::queue<T> queue_; mutable std::mutex mutex_; std::condition_variable not_empty_; };这里有几个设计细节，值得反复琢磨。
2.2 细节一：为什么把notify_one()放在锁外面
我在push里用了花括号把lock_guard的作用域限制在最小范围内，notify_one()是在锁外调用的。这个不是写代码时顺手而为，而是有实际性能考虑的。如果在持锁状态下唤醒消费者，消费者被唤醒后会立即去抢同一把锁，于是生产者和消费者在锁上产生一次无谓的争抢。虽然现代互斥锁实现会用wait_wake之类的机制缓解这个问题，但把通知放在锁外依然是更优的实践。
2.3 细节二：为什么wait必须配合 while 或带谓词的重载
std::condition_variable::wait存在一个经典陷阱——虚假唤醒。操作系统在实现线程唤醒时，不是每次都能精确控制被唤醒的线程数量，某些平台下甚至会出现"被唤醒的线程不止一个"的情况。此外，notify_all也可能导致多个等待线程同时苏醒，但队列里只有一个元素。
所以正确的写法必须是：
not_empty_.wait(lock, [this] { return !queue_.empty(); });这等价于while (!condition()) wait(lock)。只有用循环重新检查条件，才能保证即使发生了虚假唤醒，线程也不会傻乎乎地从一个空队列里取数据。这个习惯一定要刻在骨子里，换成任何语言都一样，Java 的await也要放在循环里。
2.4 细节三：析构与关闭机制的缺失
上面这个版本有个现实问题——如果消费者线程一直在pop上阻塞，而生产者再也不投递数据了，程序就永远挂在那里。实际项目里，队列经常需要显式的关闭信号。我通常会在队列里加一个closed_标志位：
void close() { { std::lock_guard<std::mutex> lock(mutex_); closed_ = true; } not_empty_.notify_all(); } std::optional<T> pop_or_none() { std::unique_lock<std::mutex> lock(mutex_); not_empty_.wait(lock, [this] { return !queue_.empty() || closed_; }); if (queue_.empty()) { return std::nullopt; } T value = std::move(queue_.front()); queue_.pop(); return value; }有了关闭机制，线程池才能优雅退出。notify_all而不是notify_one是必须的，因为关闭时需要唤醒所有阻塞的消费者，让每一个都能检查到关闭状态，正常退出循环。
锁保护队列是正确性最高、最容易推理的方案。除非你的项目有非常充分的理由去追求极致的锁竞争性能，否则生产环境首选就应该是它。C++ 标准库的std::mutex、std::condition_variable足够可靠，也没有你想的那么慢。之后所有无锁方案，都是在用复杂度换性能，得算清楚这笔账再动手。
3. 无锁队列：性能诱惑背后的内存序与 ABA 陷阱
当你在性能剖析里看到锁竞争成了热点时，就会考虑无锁队列。无锁不是说完全没有同步，而是把互斥锁的"阻塞等待"换成了"原子指令重试"。通常基于 CAS（Compare-and-Swap）或 FAA（Fetch-and-Add）完成。无锁队列的水确实很深，稍不留神就掉进内存序和 ABA 的坑里。我觉得有必要把这两个问题讲透，不然你只是抄了一份代码，跑挂了都不知道在哪排查。
3.1 CAS 循环的本质：乐观重试代替悲观等待
CAS 指令做的事，可以理解成"拿到一个内存位置的旧值，和我期望的值比一比，如果相等，就写入新值，整个比较和写入是原子的；如果不相等，什么都不做，告诉我实际值"。这套操作的英文原意是compare_exchange。
基于 CAS 的队列里，生产者的操作大致长这样：
bool push(T value) { Node* new_node = new Node(value); Node* tail = tail_.load(std::memory_order_relaxed); while (true) { Node* next = tail->next.load(std::memory_order_acquire); if (next != nullptr) { tail_.compare_exchange_weak(tail, next, ...); continue; } if (tail->next.compare_exchange_weak(next, new_node, ...)) { tail_.compare_exchange_weak(tail, new_node, ...); return true; } } }每次 CAS 失败，说明别的线程已经抢先改动了共享状态，于是重新加载最新值再试一次。这就是"乐观锁"思路，没有线程被挂起，大家都在原地打转等机会。轻负载下延迟很低，高竞争下重试会增多，但通常还是比互斥锁的上下文切换划算。
3.2 内存序：比锁难理解一万倍的正确性门槛
无锁编程里最诡谲的部分不是 CAS 本身，而是内存序。你写代码时知道变量 A 先改、变量 B 后改，可 CPU 和编译器为了性能会乱序执行。好在 C++ 提供了一套 memory order 让你约束这个乱序：
memory_order_relaxed：只保证原子性，不保证顺序memory_order_acquire：本线程后续的读，不能被重排到这次读之前memory_order_release：本线程之前的写，不能被重排到这次写之后memory_order_seq_cst：全局顺序一致，最强约束，性能最差
在无锁队列里，最常见的是把"发布新节点"这一步做成 release，把"获取新节点"这一步做成 acquire。比如生产者写入节点数据后用 release 发布指针，消费者用 acquire 读取指针。这样，消费者一旦拿到了指针，它就能保证看到生产者发布前写入的所有数据。如果这里图省事全部用 relaxed，那你拿到的就是把环环相扣的并发逻辑变成了盲盒，编译器和硬件都帮不了你。
记忆方法很简单：release 负责把数据"送出去"，acquire 负责把数据"接下来"。配对使用，数据才能安全旅行。单独一个 relaxed 就是裸奔。
3.3 ABA 问题：为什么"内存地址没变"会骗过你
无锁队列还有个让人头皮发麻的问题叫 ABA。设想一个无锁栈或队列，某线程 A 读取到指针 P，正准备 CAS 把 P 换成新值 Q。此时线程 B 介入，把 P 删除、释放，又新建了一个节点 R，恰好操作系统把 R 分配到了和 P 相同的内存地址。这时线程 A 的 CAS 拿旧值 P 进行比较，发现地址一样，以为状态没变，实际上整个链表的拓扑已经变了好几轮，于是成功写入了一个已经失效的指针。
落进 ABA 的真实后果通常不是报错，而是静默内存错误，链表悄悄断开，某个线程读到了损坏的数据。应付 ABA 的经典招数包括：
- 延迟回收（Hazard Pointer）：不立即释放内存，等所有线程都不持有该指针了再释放。
- 带计数器的原子指针：用
std::atomic<std::pair<void*, uint64_t>>包装指针和版本号，每次修改都让计数加一，CAS 时同时比较指针和版本号。 - 环形数组：广播式队列用固定内存槽位，不涉及节点删除和复用，天然规避 ABA。
我见过不少团队把无锁队列封装得漂漂亮亮，上线之后在极端流量下偶发数据错乱，排查到最后都是 ABA。所以在选无锁方案前，先回答自己一个很诚实的问题：你真的有足够时间做对了吗？如果没有，锁队列就是更好的方案。
4. 阻塞、非阻塞、有界、无界：按场景选型才不踩坑
实现一个线程安全队列只是第一步，更关键的决策在于你需要在什么场景下用什么样的队列。很多人以为线程安全队列就是一把锁加一个std::queue完事，实际上按阻塞行为、容量限制组合起来，至少有四类用法，各自的坑也不一样。
4.1 四象限选型模型：先问自己是哪一类
我习惯用一个四象限来给队列定性：横轴是阻塞/非阻塞，纵轴是有界/无界。
- 有界阻塞队列：消费者没数据就休眠，生产者放不进就阻塞或等待。这是线程池任务队列最常用的形态。因为容量固定，写满了会让生产者停下来，天然形成背压。
- 有界非阻塞队列：放不进就返回失败，取不到就返回空。适合网络收包、硬件采集这类"你不收我丢，我不等你"的实时场景。
- 无界阻塞队列：只限制消费端等待，生产端永远能放进去。适合生产者偶尔突增、消费者吞吐足够稳定的场景。缺点是如果生产端暴走，内存会被无限撑爆。
- 无界非阻塞队列：生产端永远成功，消费端拿不到就返回。这种组合我在日志采集里用得比较多，因为不能因为日志量少就让业务线程去睡觉，也不能因为日志量多就让业务线程被阻塞，只能尽量缓存，消费端尽力消费。
4.2 有界队列的背压：保护系统不被冲垮的关键
真正值得多说几句的是有界队列。很多人一想到 "线程安全队列" 就顺手写了个无界的，觉得反正内存很大，无所谓。可一旦某个下游组件变慢，消息会在队列里堆积，最终耗尽内存，进程直接 OOM。有界队列的意义在于，它把"队列本身"变成了一个信号灯，让生产端的调用方能够感受到下游的真实压力，从而触发节流、降级或快速失败策略。
实现有界队列时，push的阻塞版本应该等待"不满"这个条件，配合另一个条件变量：
void push(const T& value) { std::unique_lock<std::mutex> lock(mutex_); not_full_.wait(lock, [this] { return queue_.size() < capacity_ || closed_; }); if (closed_) throw QueueClosed(); queue_.push(value); not_empty_.notify_one(); }如果你不需要阻塞，可以提供一个try_push接口，满员时直接返回 false。背压的哲学既不是"硬扛"，也不是"丢弃"，而是"从源头放慢速度"。
4.3 实际工程中的典型组合
我在实际项目中用得最多的是两种组合。
第一种是线程池任务队列，做成有界阻塞队列。容量通常设为thread_count * 2或2^N这样的值。任务积压会让提交任务的业务方感受到阻塞从而自然降速，不会无限增长内存。
第二种是异步日志队列，做成有界非阻塞队列，满了就直接丢弃低级别日志，把内存安全放在第一位。日志丢失一点没关系，进程不能挂，这是优先级排序的问题。
如果你的场景对吞吐量要求极高但正确性问题不算太敏感（比如只做统计消息的 flavor），可以考虑批量提交：生产端攒一批再一次性 push 进队列，显著减少同步次数。对每个小任务都加锁的写法，在高频小消息场景下分分钟锁竞争爆炸。
5. 实测与排错：从隐含竞态到伪共享的性能教训
队列写完之后怎么验证它对不对，可能是比实现队列更考验功力的事。这里我不聊复杂的并发验证理论，就说我踩过的三个比较典型的坑：隐藏的竞态、伪共享导致的性能异常，以及测试环境速度差带来的假阳性。
5.1 一个藏了半年才发现的数据竞争
有段时间我维护一个内部消息队列，功能非常简单，加锁保护的，逻辑上看起来天衣无缝。但偶发出现消费线程读到了脏数据。后来才定位到问题根本不在队列内部，而是队列的某个使用方在外部拷贝元素时，元素本身内部有裸指针。队列只保护了队列结构，并没有深拷贝元素内容。另一个线程通过别的方式修改了元素内部的数据，于是队列的同步白白做了。
这件事给我一个教训：线程安全的队列，不等于放在队列里的对象也是线程安全的。如果你的元素是智能指针、裸指针或者含有共享可变状态的对象，一定要明确所有权边界，或者干脆用值语义（拷贝或移动）传入队列。
5.2 伪共享：一个常见的性能杀手
另一个典型坑是伪共享。很多无锁队列为了性能，会把头指针和尾指针放在两个缓存行里。如果这两个变量恰好相邻或共用一个缓存行，CPU 缓存一致性协议（MESI 协议）会在不同核心修改它们时不断失效整个缓存行。头指针更新一点儿，尾指针所在的核心就要重新加载该缓存行，反过来也一样。这导致并发吞吐不升反降，甚至比加锁还慢。
解决办法就是缓存行填充：
struct alignas(64) AlignedAtomic { std::atomic<size_t> value; uint8_t padding[64 - sizeof(std::atomic<size_t>)]; };在极致的性能测试里，我把一个队列加上缓存行填充后，吞吐量提升了接近三倍。这个提升不是玄学，而是避免 CPU 核心之间互相拖后腿。当然，alignas(64)在现代 x86 上非常有效，ARM 平台的缓存行通常是 64 或 128 字节，做嵌入式端的同学得留意。
5.3 测试与基准：如何写一个靠谱的并发验证
最后说一点关于测试方法的心得。多线程 bug 是概率问题，单次跑过不代表安全。我习惯写一个蛮力并发测试：
// 生产者生产 N 个递增编号，消费者校验编号是否为乱序或缺失 constexpr int kProducers = 4; constexpr int kConsumers = 4; constexpr int kItemsPerProducer = 100000;理论上消费者收到的数据应该是一组从 0 到 N-1 的不重复编号，任何重复、缺失、乱序都说明队列出了问题。配合-fsanitize=thread或者 Java 里类似的竞态检测工具跑上一整夜，再随机模拟偶尔的调度延迟，很多问题就能暴露得比较快。
如果你用的是无锁实现，ASE 里建议再加上内存序检测工具，看看是否有哪个共享的指针被用错了 memory order。这种工具检查出的问题很难靠人眼看出来，必须依赖机器。
我想说的其实就一句话：与其纠结一个万能队列，不如先想清楚自己的场景是重吞吐还是重可靠，然后选择锁或者无锁，把一两个队列模型做到极致，比收集各种花哨的实现要有价值得多。每次我从无锁的坑里爬出来再回到锁队列时，都会想起一个行业前辈说的话——先保证正确性，再考虑优化；如果没有正确性，优化就是反向伤害。
暴力到O(n)高效解)
