写多线程代码的人,大概率撞过这样的场景:一个线程在往栈里写数据,另一个线程在取数据,跑着跑着要么段错误,要么数据对不上。最省事的办法是加 mutex,但临界区又短又热的时候,锁竞争就成了新的瓶颈。于是大家把目光投向“无锁”:用一个原子指针和一段 CAS 循环,能不能让栈在多个线程之间安全流转?这篇文章就把这个思路拆开讲清楚——底层原理、C++ 代码怎么落地、以及真正写进工程之前绕不开的那几个大坑。无论你是刚接触std::atomic的新手,还是写过不少锁代码但想理解无锁编程的老手,都能从里面找到点东西。
我先把结论放在前面:无锁不是“免费的午餐”,它只是把“阻塞”换成了“重试”,用一个更复杂的算法模型换掉了操作系统调度。CAS(Compare-And-Swap)是这个模型的核心原语,而 Treiber 栈是这个模型最经典的教学案例。下面我们从头开始,一层层把它剥开。
1. 从一次栈竞争说起:为什么要把“锁”拿掉
1.1 锁的问题不在“锁”本身,而在你锁了一个非常短的临界区
很多人一提到多线程共享数据结构,第一反应就是加锁:
std::mutex m; std::vector<int> stack; void push(int v) { std::lock_guard<std::mutex> lk(m); stack.push_back(v); }这没问题,绝大多数场景下它就是最正确的选择。但如果你用 profiler 去抓热点,会发现一种典型的“锁瓶颈”长这样:栈操作的临界区本身只有几行代码,几十条指令就能跑完,可是线程之间为了抢这把锁,互相让出 CPU、又互相唤醒,调度开销和缓存失效的开销加起来,比临界区本身的执行时间还大。
锁的两个本质代价是绕不开的。第一,阻塞会触发操作系统介入,线程切换、唤醒、排队,这些动作都是微秒级的,而一次 push 本身可能只需要几十纳秒;第二,锁意味着写者之间要串行化,拿到锁的线程才能碰数据,其他线程哪怕只在旁边读,也得排队。更要命的是优先级反转这类问题,在锁模型下天然存在。
所以“把锁拿掉”的动机,本质上不是锁不好,而是临界区太短、竞争太频繁。此时我们希望:每个线程想做 push/pop 时,直接尝试原子操作,成功了就干,失败了就重试,不给操作系统机会把自己挂起。
1.2 无锁栈能承诺什么,不能承诺什么
无锁栈通常指的是基于 CAS 实现的链式栈,学术上一般叫 Treiber 栈。它承诺的是:任意时刻,至少有一个线程能在有限步内完成操作,系统整体不会因为某个线程被调度走而永久卡死。换句话说,它的“进度保证”是 lock-free 级别的——比阻塞算法强,但不是绝对无等待(wait-free)。
但它不能承诺什么?不能承诺“每个线程都很快”。当大量线程同时竞争栈顶指针时,CAS 失败的线程会反复读取并重试,极端情况下可能出现活锁或者性能雪崩。也不能承诺“一定比 mutex 快”。无锁栈把锁竞争变成了缓存行竞争,多个线程同时对一个原子变量做 CAS,会让那个缓存行在核间疯狂来回飞,代价同样不低。
一个比较务实的认知是:无锁栈适合临界区极短、线程数可控、对实时性有要求的场景;它解决的问题是“不阻塞”,而不是“更快”。有了这层认知,我们再往下看实现细节,就不容易被花架子带偏。
2. Treiber栈的骨架:一个原子指针加两段CAS循环
2.1 为什么栈天然适合“单指针”表示
数组栈的问题是动态扩容和空洞管理。你用std::vector当栈,push 需要处理 size 增长时的 realloc,pop 之后还得决定元素是物理删除还是逻辑删除;这些操作一旦并发起来,就不只是栈顶一个变量的问题了。
链式栈就清爽很多。每个元素是一个独立节点,节点里有值和 next 指针,整个栈只需要维护一个头指针head_。push 就是往 head 前面插入节点,pop 就是从 head 摘走第一个节点。栈的所有并发状态都浓缩在head_这一个指针上,天然适合用一个std::atomic<Node*>来表达。
这里有一个很多人第一次看会困惑的点:为什么不用环形缓冲区加原子索引?那也是一种无锁结构,但它在“容量固定、pop 只读下标”的语义下才成立。真正的栈你希望节点能动态分配、能无限压入,链表结构就比环形数组灵活得多。Treiber 栈能把全部复杂度收敛到一个指针上,靠的就是链表这个形态。
2.2 push循环:第一次失败不是坏事,expected会帮你兜底
先看 push 的最基础版本:
template <typename T> class LockFreeStack { public: struct Node { T value; Node* next = nullptr; }; void push(T value) { Node* node = new Node{std::move(value), head_.load(std::memory_order_relaxed)}; // 注意 compare_exchange_weak 的 expected 是引用,失败后会自动更新 while (!head_.compare_exchange_weak(node->next, node, std::memory_order_release, std::memory_order_relaxed)) { // node->next 已经被更新为最新的 head,直接重试 } } Node* pop_raw(); // 稍后讲 private: std::atomic<Node*> head_{nullptr}; };push 的逻辑一句话就能说清:先把新节点的 next 指向当前栈顶,再用 CAS 把栈顶替换成新节点。但这里藏着两个细节。
第一个细节,compare_exchange_weak的第一个参数是 expected,它以引用的方式传入。第一次执行时我们把node->next当作 expected 传进去,它等于我们刚从head_读到的旧栈顶。如果 CAS 失败,说明在“读旧栈顶”和“尝试写新栈顶”之间,已经有别的线程修改了head_;此时库会把 expected 更新成最新的head_值,也就是node->next会被自动改写。所以我们不需要在循环里再手动 load 一次,直接重试比较即可。
第二个细节,为什么用compare_exchange_weak而不是compare_exchange_strong。weak 版本在满足比较条件时也可能“假失败”,这在循环里无所谓,多转一圈就回来了;但它在某些平台上能生成更高效的指令,尤其是 ARM 上的 LDXR/STXR 循环。无锁循环几乎总是应该选 weak。
2.3 pop循环:读节点内容之前,先确认栈顶还是你看到的那个
pop 的逻辑稍微绕一点:
Node* pop_raw() { Node* old_head = head_.load(std::memory_order_relaxed); while (old_head && !head_.compare_exchange_weak(old_head, old_head->next, std::memory_order_acquire, std::memory_order_relaxed)) { // old_head 失败后被自动更新,继续尝试 } return old_head; // 空栈时返回 nullptr }流程是:读当前栈顶,把它的 next 当作“目标新栈顶”,用 CAS 尝试把head_从 old_head 换成 old_head->next。如果成功,old_head 就是被弹出的节点,返回给调用者。
为什么循环条件还要判断old_head?因为空栈时 load 出来是 nullptr,old_head->next不能解引用,所以必须短路。
为什么 CAS 失败后 old_head 需要自动更新?因为失败意味着栈顶已经不是我们读到的那个节点了,要么被别的线程 pop 了,要么被 push 插入了新节点。此时再用旧 old_head 的 next 做期望值没有意义,必须拿到最新的 head 重新来。
这里有个非常重要的隐藏问题:参数里的old_head->next是在调用 CAS 前求值的。如果 old_head 指向的节点已经被别的线程弹出去并且释放掉内存,那这里就是典型的 use-after-free。这段代码目前能跑,完全依赖“别乱释放”这个约定。我们在下一节专门处理它。
2.4 给一份能真正安全跑起来的教学版代码
结合上面两部分,给一个可以实际编译运行的简化版。为了把 CAS 逻辑讲清楚,这个版本的 pop 返回裸节点指针,节点的 delete 由调用者统一负责,绝不在 pop 的瞬间释放:
#include <atomic> #include <utility> template <typename T> class LockFreeStack { public: struct Node { T value; Node* next = nullptr; }; void push(T value) { Node* node = new Node{std::move(value), head_.load(std::memory_order_relaxed)}; while (!head_.compare_exchange_weak(node->next, node, std::memory_order_release, std::memory_order_relaxed)) { } } Node* pop_raw() { Node* old_head = head_.load(std::memory_order_relaxed); while (old_head && !head_.compare_exchange_weak(old_head, old_head->next, std::memory_order_acquire, std::memory_order_relaxed)) { } return old_head; } private: std::atomic<Node*> head_{nullptr}; };调用方的约定是:pop_raw 返回的节点,先收进一个“退役列表”,等所有线程都退出后再统一 delete。这样节点存活时间覆盖了所有线程的访问窗口,不会出现悬空指针。这个约定不优雅,但它让 CAS 的核心逻辑暴露得干干净净。
有的读者会问:那你这个栈的普通用法呢?直接用Node*操作,多别扭。确实别扭,但这正是无锁栈工程化必须迈过的一道坎:节点回收和 ABA 问题不解决,就不可能做出一个“能正常释放内存”的 pop。下一节就把这道坎掰开揉碎讲清楚。
3. 无锁栈的阴影:ABA问题与节点回收其实是一件事
3.1 ABA问题是怎么从“地址巧合”里长出来的
很多教程说 ABA 问题就是“值从 A 变成 B 又变回 A,CAS 误以为没变过”。这个说法太抽象了,栈这个场景里它长这样。
线程 1 读取栈顶,发现 head 指向节点 A,它准备执行 CAS,把 head 从 A 改成 A 的 next(假设是 B)。就在这个瞬间,线程 2 抢先 pop 了 A,head 变成 B;然后线程 2 又 push 了一个新节点。这个新节点如果恰好复用了 A 的内存地址,那 head 的地址值又变回了“和 A 相同的数字”。
线程 1 的 CAS 一看,expected 是 A 的地址,现在 head 的地址值恰好等于它,比较通过,于是成功把 head 写成了 B。问题来了:现在的栈顶其实是一个全新的节点,它的 next 和当年的 A 完全不同,但线程 1 强行把它后面的指针指向了 B。结果就是链表结构被打乱,轻则丢失节点,重则形成环。
ABA 出现的根因是什么?是“地址值相同”和“结构未变”被混为一谈了。而要触发这个巧合,几乎总是因为你释放了节点 A,之后又分配到了同一个地址。如果你从不释放任何节点,new 出来的指针不可能是旧地址,ABA 就不会发生。这也是为什么上一节的教学版能安全跑:它把释放推迟到了最后,从根上掐掉了 ABA 的引线。真实工程里你不可能永远不释放内存,所以必须用别的手段。
3.2 带版本号的16字节CAS:算法原理与代码
解决 ABA 最直接的办法:不要只比较一个指针,而是比较“指针加版本号”。每次有人成功改动了栈顶,版本号就加一。即使指针值回到原来的地址,版本号对不上,CAS 也会失败。
C++ 层面可以这样设计一个复合载荷:
#include <atomic> #include <cstdint> struct TaggedNode { Node* ptr; std::uint64_t tag; }; std::atomic<TaggedNode> head_{{nullptr, 0}};std::atomic<TaggedNode>在 x86-64 平台上通常会被编译成 16 字节的原子比较交换指令cmpxchg16b。注意“通常”这两个字,标准并不保证这一点,所以工程上必须加上static_assert(head_.is_always_lock_free)这类检查,否则它退化成锁,无锁就名存实亡了。
push 变成这样:
void push(T value) { Node* node = new Node{std::move(value), nullptr}; TaggedNode old = head_.load(std::memory_order_relaxed); while (true) { node->next = old.ptr; TaggedNode desired{node, old.tag + 1}; if (head_.compare_exchange_weak(old, desired, std::memory_order_release, std::memory_order_relaxed)) { break; } // old 在这里会被自动更新成最新的 head 和 tag } }pop 同理,读取head_得到{ptr, tag},准备把它改成{ptr->next, tag+1},CAS 同时比较指针和版本号。只要中间发生过哪怕一次 push/pop,版本号就会变化,ABA 就被挡在门外。
这个方案的优点是思路简单,缺点也明显:依赖 16 字节 CAS,在 ARM 平台未必原生支持,而且它只解决了 ABA,没有解决“节点被释放后其他线程还在读”的生命周期问题。你依然需要一个内存回收策略。另一个变体是把版本号塞进指针的低位 bit,因为节点地址通常是对齐的,低两位必然为 0。这种做法省掉了 16 字节比较,但版本号位数有限,很容易溢出,实际使用时要配合回收周期设计,工程复杂度更高。
3.3 Hazard Pointer:工程上更完整的回收方案
Hazard Pointer(危险指针)是目前无锁数据结构里最常见的回收方案之一。它的核心思想是:每个读者线程只会短暂地“占用”一个节点——从读 head 到 CAS 成功的这段窗口。我们让每个线程维护一个公开的 slot,记录自己当前正在读的节点指针。写者在回收节点之前,先去所有线程的 slot 里查一遍,如果有人还指着这个节点,就暂时不回收,把它放进待回收列表,等下一轮再检查。
看起来简单,落地时需要处理几个细节。slot 本身要原子读写;读者在 load head 之后、使用节点之前,要把节点地址发布到自己的 slot,并且重新读一次 head 确认没有变化;写者扫描 slot 和回收节点之间还有窗口,所以常常会把节点先放进一个“退役队列”,等下一轮垃圾回收时才真正 delete。经典实现可以参考算法论文里的“retire list + scan”流程,工程库也有现成的封装。
Hazard Pointer 的优点是回收相对及时、内存占用可控,缺点也很明显:读者每次操作要多写一次 slot,多读一次 head,多了一道工序;写者回收时要扫描所有线程,线程越多开销越大。它在 8 线程以内往往表现不错,线程数几十上百时,扫描成本就会让人头疼。
另一条路线是 epoch-based reclamation(基于纪元的回收)。系统维护一个全局纪元,读者进出临界区时更新线程本地纪元,写者将退役节点挂到当前纪元的列表里,等所有线程都进入新纪元后,再统一释放旧纪元的节点。实现比 Hazard Pointer 简单,但内存延迟释放的峰值更高。各有取舍。
3.4 不要被“教学demo能跑”骗了
我见过不少人把 Treiber 栈的教学代码直接抄进生产环境,然后三天两头出诡异崩溃:某天 push 的节点神秘消失,某个线程读到了刚被 free 的 next,甚至死循环。这些都是同一个病根——把“演示期不释放”当成了“永远不需要回收”。
请记住一个判断标准:如果一个无锁栈的 pop 里直接写delete node,而 push 里直接new Node,那这个栈十有八九是有问题的。节点释放“太早”,会让其他线程踩到悬空指针;节点释放“太晚”,内存会无限膨胀。真正要解决的是“什么时候释放是安全的”,而不是“怎么释放最快”。CAS 只负责把栈顶指针改对,它管不了指针指向的内存是否还活着,这是两套完全不同的机制,缺一不可。
4. 内存序实战:为什么是release/acquire而不是relaxed
4.1 relaxed、release、acquire三兄弟的直觉
std::memory_order是 C++ 给原子操作指定的同步强度。很多人觉得它玄,其实用三个词就能建立直觉。
relaxed 只保证原子操作本身的不可分割性,不保证任何其他内存操作在它前后的顺序。你可以把它理解为“我只需要这个变量的读写是原子的,其他随便”。
release 是针对写操作的语义:如果一个线程执行 release 写,那么它之前的所有普通内存写,都会在这次 release 写之后对其他线程可见。关键词是“之前的所有写”。
acquire 是针对读操作的语义:如果一个线程执行 acquire 读,那么它之后的所有普通内存读,都能看到某个 release 写之前发布的数据。关键词是“之后的读”。
release 和 acquire 一旦配对,就建立了一条“跨线程 happens-before”的链。这条链让数据从写线程传递到读线程时,顺序是确定的,不会出现“head 已经指向新节点,但新节点的 value 还没写进去”这种恐怖状况。
4.2 push用release,pop用acquire,理由拆开给你看
回到我们的代码。push 里,一个线程做了两件事:先把 value 写进新节点,再通过 CAS 把新节点挂到 head。如果 CAS 用的是 relaxed,会怎么样?在弱内存序的 CPU 上,另一个线程可能看到“head 已经指向新节点”,但读取新节点的 value 时却看到旧值、甚至半更新的值。因为 value 的写入没有被 release 语义“发布”出去。
所以 push 的 CAS 成功序必须用memory_order_release。它保证:一旦新节点对全世界可见,节点的 value 和 next 一定已经完整写好了。
pop 里同样有跨线程的读依赖。线程从 head 摘下一个节点,然后去读它的 value,这个 value 是当初某个 push 线程写的。CAS 成功序用memory_order_acquire,就能确保在获取 head 之后,读取节点 value 时能看到 push 线程 release 之前的所有写入。这就是 release/acquire 配对的含义。
那 pop 的 CAS 也是写操作(把 head 改为 next),为什么不用acq_rel或者 release?因为 pop 成功之后,它没有额外发布什么数据给其他线程。它只是拿走了一个节点,这个节点被摘走后,后续操作由调用者自己负责;如果调用者想把这个节点再放回栈里,那是下一次 push 的事,会由 push 的 release 来处理。所以单看 pop 这一步,acquire 的读语义已经覆盖了它需要的同步,给它 acq_rel 属于过度设计。不过你要是图省事统一写成acq_rel,正确性也不会受太大影响,只是白付一点代价。
4.3 compare_exchange_weak的两个内存序参数怎么填
compare_exchange_weak有四个参数:expected、desired、成功序、失败序。很多人搞不清失败序有什么用,这里顺手说明白。
CAS 成功意味着原子变量确实被写成了 desired,这时需要按成功序去释放或获取数据,所以成功序是重头戏。CAS 失败意味着原子变量没有被修改,当前线程只是“看一眼”它的值,没有发生数据交换,也就不需要 release 的发布作用;此时用memory_order_relaxed就够了,避免在失败路径上引入不必要的内存栅栏。
在 x86 平台上,这点差别几乎感觉不出来,因为 x86-TSO 模型本身就比较强,release/acquire 通常不额外生成栅栏指令。但在 ARM 和 RISC-V 上,错误地用 acquire 做失败序会让失败重试路径也付出昂贵的同步代价,无锁栈在高竞争环境下会雪上加霜。写成success=release, failure=relaxed(push 场景)或success=acquire, failure=relaxed(pop 场景),是兼顾正确性和性能的标准做法。
还有一个容易被忽略的点:compare_exchange_weak和compare_exchange_strong的语义差别。strong 保证“值不一样就一定失败”,weak 允许“值一样时也偶尔失败”。在循环里应该用 weak,它能带来更好的指令生成;在“只尝试一次”的场景里必须用 strong,否则一个假失败就可能让逻辑走错分支。在不同的 CPU 上,指令实现的差异比想象中大,正确选型能直接吃到硬件红利。
5. 验证、压测与真正的工程结论
5.1 正确性验证:计数等式加TSAN
写完无锁代码,第一件事不是看性能,而是验证正确性。一个直观的验证方法是:让 N 个线程并发做固定次数的 push 和 pop,然后用原子计数器记录总数。等所有线程结束后,统计栈里剩余的节点个数,验证下面的等式:
push 总数 = pop 成功总数 + 栈内剩余节点数
如果这个等式不成立,说明存在节点丢失或者被重复弹出。等式成立不代表一定正确,但等式不成立一定有问题。
更强的验证手段是用 ThreadSanitizer(TSAN)跑一遍同样的负载。TSAN 能检测数据竞争,包括原子操作和普通读写之间的冲突。编译时加上-fsanitize=thread,跑出来的报告如果干净,至少说明没有裸奔的数据竞争。注意 TSAN 打开以后执行速度会慢很多,迭代调试时把循环次数调小一点。
我惯用的压测骨架是这样:
#include <atomic> #include <thread> #include <vector> #include <chrono> constexpr int kThreads = 8; constexpr int kOps = 200000; std::atomic<long long> total_push{0}; std::atomic<long long> total_pop{0}; LockFreeStack<int> stack; void worker(int seed) { std::vector<LockFreeStack<int>::Node*> retired; for (int i = 0; i < kOps; ++i) { stack.push(seed * 1000000 + i); total_push.fetch_add(1, std::memory_order_relaxed); if (auto* node = stack.pop_raw()) { retired.push_back(node); total_pop.fetch_add(1, std::memory_order_relaxed); } } for (auto* node : retired) delete node; } int main() { auto start = std::chrono::steady_clock::now(); std::vector<std::thread> threads; for (int t = 0; t < kThreads; ++t) threads.emplace_back(worker, t); for (auto& th : threads) th.join(); auto end = std::chrono::steady_clock::now(); int remaining = 0; for (auto* cur = stack_head_approx(); cur; cur = cur->next) ++remaining; // 验证 total_push == total_pop + remaining,然后打印耗时 }这里的stack_head_approx()是为了在测试结束后遍历栈内节点写的辅助函数。注意每个线程弹出的节点都进了自己的 retired 列表,等线程退出后统一 delete,这样并发访问期间这些节点永远不会被真正释放,避免悬空指针问题。
5.2 压测结果到底怎么看:无锁也可能更慢
压测数据很容易让人自嗨,但解读时一定要冷静。我拿 8 线程各自 push/pop 20 万次测过,单纯看栈操作耗时,无锁版能比 mutex 版快不少,量级上大概是“明显更快”而不是“快一点”。但这里有个陷阱:上面的测试里,每个线程预先把节点放进了 retired 列表,没有把 new/delete 计入栈操作。如果把节点分配和释放的成本算进来,无锁版的优势就会大幅缩水,甚至反转。
为什么?因为new和delete本身就是全局互斥的,多个线程频繁 new/delete 会产生新的锁竞争。真实工程里要想充分发挥无锁栈的优势,通常要配合节点池或上文提到的 Hazard Pointer 回收机制,把分配成本从热点路径上挪走。也就是说,你最终得到的“快”,是“无锁 CAS + 池化分配 + 延迟回收”三者合力的结果,不能全记在 CAS 头上。
还有一个更反直觉的现象:线程数很低时,mutex 版不一定比无锁版慢,甚至可能更快。因为未争用的 mutex lock/unlock 开销很小,而无锁版每次操作都要 load head、写 next、做 CAS,代码路径更长。线程数很高且竞争激烈时,CAS 反复失败,缓存行在核间疯狂迁移,性能也会急剧下降。无锁栈的性能曲线不是单调向好的,它更像一个峰顶之后急速下坠的抛物线。真正优势明显的区间,是“临界区极短、线程数中等、操作频繁且不愿被阻塞”的那段窗口。
5.3 什么场景千万别用无锁栈
我见过的无锁栈翻车案例,几乎都栽在场景误判上。
第一种,临界区里干了太多活。有人把序列化、日志、内存拷贝全塞进 push/pop 里,然后大喊无锁不靠谱。这冤枉了无锁,真正的问题是临界区太长,CAS 重试成本跟着变大。临界区一长,老老实实用 mutex 更划算。
第二种,需要严格 FIFO 语义的地方用了栈。栈是 LIFO,这是数据结构层面的语义,跟锁不锁没关系。无锁栈不能给你队列的公平性,消费者可能老读到最新数据而饿死底部的老数据。
第三种,团队没有无锁编程经验,又把它用在核心链路上。无锁代码难写更难维护,稍不留神就是内存泄漏或者悬空读。线上问题随机复现的时候,你很难向老板解释“ABA 问题导致了一个神秘的偶发故障”。如果 mutex 版本一天才 10 微秒的竞争开销,而无锁版本可能要你花两周去调回收算法,投入产出比真不一定划算。
第四种,节点数量不可控、内存回收困难。无锁栈的节点可能被任意线程延迟释放,如果压测阶段就出现大数据量的峰值,退役列表可能堆积大量节点,内存峰值比锁版本高出几倍。没有充分压测之前,别轻易把内存模型复杂化。
我个人现在写并发代码的原则很简单:默认加锁,profile 打脸了再上无锁;而无锁方案里,优先考虑 mutex 退避、读写分离、批量提交这些更“粗粒度”的优化。无锁栈更多时候是我的思维训练场——它逼我把原子语义、内存序和生命周期想得明明白白。等你想明白了,再回头去看那些 lock-free 队列、Hazard Pointer 实现,会发现所有问题都能归到“指针怎么改”和“内存什么时候能还”这两件事上。