4. 无锁编程:无锁栈与队列、ABA问题与解决方案、内存回收机制

无锁编程,说白了就是不用互斥锁,也能保证多线程安全。很多刚接触并发的朋友一听“无锁”就两眼放光,觉得性能肯定起飞。嗯,确实能起飞,但坠机的概率也不小。

我个人习惯把无锁编程看作“在刀尖上跳舞”。它利用的是硬件级别的原子操作——CAS(Compare-And-Swap)或者 LL/SC。你想想看,没有操作系统调度器的介入,全靠 CPU 指令保证一致性,这效率能不高吗?

但代价是什么?是极其复杂的逻辑和几个非常容易踩的坑。今天我们就来聊聊无锁栈、无锁队列,以及那两个臭名昭著的问题:ABA 和内存回收。

4.1 无锁栈:从 Treiber Stack 说起

无锁栈最经典的实现就是 Treiber Stack。它的核心思想很简单:用 CAS 操作来替换链表的头指针。

伪代码大概长这样:

push(value):
    node = new Node(value)
    loop:
        node.next = head
        if CAS(&head, node.next, node):
            break

pop():
    loop:
        old_head = head
        if old_head == NULL:
            return EMPTY
        new_head = old_head.next
        if CAS(&head, old_head, new_head):
            return old_head.value

逻辑上很清晰对吧?push 就是把新节点的 next 指向当前 head,然后用 CAS 把 head 更新成新节点。如果 CAS 失败,说明有其他线程改了 head,那就重试。pop 同理。

我的经验: 写 Treiber Stack 的时候,最容易被忽略的是 ABA 问题。我刚开始写的时候,觉得这代码天衣无缝,结果跑起来数据全乱套了。后面会详细讲。

4.2 无锁队列:Michael-Scott 队列

栈是 LIFO,相对简单。队列是 FIFO,需要同时操作 head 和 tail,复杂度直接翻倍。业界最经典的无锁队列是 Michael 和 Scott 在 1996 年提出的。

它的核心思路是:用一个 dummy 节点(哨兵节点)来简化边界条件。队列初始化时,head 和 tail 都指向这个 dummy 节点。

enqueue(value):
    node = new Node(value)
    loop:
        tail = this.tail
        next = tail.next
        if tail == this.tail:  // 检查 tail 是否被修改
            if next == NULL:
                if CAS(&tail.next, NULL, node):
                    CAS(&this.tail, tail, node)  // 尝试更新 tail
                    break
            else:
                CAS(&this.tail, tail, next)  // 帮助其他线程推进 tail

dequeue():
    loop:
        head = this.head
        tail = this.tail
        next = head.next
        if head == this.head:
            if head == tail:
                if next == NULL:
                    return EMPTY
                CAS(&this.tail, tail, next)  // 队列正在 enqueue,帮忙推进 tail
            else:
                value = next.value
                if CAS(&this.head, head, next):
                    return value

你可能会问:为什么 enqueue 里要帮别人推进 tail?这就是 Michael-Scott 队列的精妙之处。当一个线程成功插入节点但还没来得及更新 tail 时,另一个线程发现 tail 落后了,就主动帮它更新。这种“互助”机制保证了队列不会卡死。

注意: 无锁队列的 enqueue 和 dequeue 操作中,CAS 的 ABA 问题同样存在。而且因为涉及两个指针(head 和 tail),ABA 的表现形式更隐蔽。

4.3 ABA 问题:一个经典的陷阱

什么是 ABA 问题?我用一个场景来解释。

假设线程 A 要 pop 栈顶元素。它读取了 head 指向节点 X,然后准备 CAS。但在它 CAS 之前,线程 B 把 X 和 Y 都 pop 了,然后又 push 了一个新节点 Z,巧合的是 Z 的地址恰好和 X 一样(被内存回收后重新分配了)。

这时候线程 A 的 CAS 检查 head 的值,发现还是那个地址,就认为 head 没变,于是 CAS 成功。但实际上,整个栈的结构已经变了,head 指向的已经不是原来的 X,而是全新的 Z。数据就乱了。

为什么会这样?因为 CAS 只比较“值”,不比较“版本”。

步骤 线程 A 线程 B 栈状态 (head → ...)
1 读取 head = 0x100 (节点 X) — 0x100 → 0x200 → NULL
2 — pop X, pop Y NULL
3 — push Z (地址 0x100) 0x100 → NULL
4 CAS(&head, 0x100, 0x200) — CAS 成功,但 0x200 早已被释放

这就是 ABA。地址 A 被改成 B,又改回 A,CAS 就傻傻分不清了。

4.4 解决方案:标记指针与双 CAS

解决 ABA 最直接的办法是:给指针加个“版本号”。每次修改指针时,版本号一起递增。这样即使地址复用,版本号也对不上。

在 32 位系统上,可以用 double-word CAS(DCAS)把指针和计数器打包在一起。64 位系统更简单,直接用 128 位的 CAS(比如 cmpxchg16b 指令)。

// 伪代码:带标记的指针
struct tagged_pointer {
    void* ptr;
    uint64_t tag;
};

push(value):
    node = new Node(value)
    loop:
        old_head = head
        node.next = old_head.ptr
        new_head = {node, old_head.tag + 1}
        if DCAS(&head, old_head, new_head):
            break

我建议在实现时,优先考虑用 128 位 CAS。因为 DCAS 在 x86 上不是所有 CPU 都支持,而 cmpxchg16b 从 Nehalem 架构开始就普及了。

核心要点: 解决 ABA 问题的本质是让 CAS 能区分“同一个地址的不同生命周期”。版本号是最优雅的方式。

4.5 内存回收:无锁编程的另一个大坑

无锁编程里,谁负责释放节点?这是个要命的问题。

你想想看,线程 A 从栈里 pop 了一个节点,正准备读取它的值。结果线程 B 也看到了这个节点,把它 free 掉了。线程 A 再去读,就是野指针,程序直接崩溃。

所以,无锁编程的内存回收不能像普通程序那样“用完就 free”。必须保证:没有任何线程还在访问这个节点时,才能释放它。

业界主要有两种方案:Hazard Pointer 和 RCU。

4.5.1 Hazard Pointer(风险指针)

Hazard Pointer 的思路很朴素:每个线程维护一个“风险指针”列表,里面记录它当前正在访问的节点地址。其他线程在释放节点前,先检查这个节点是否在某个线程的风险指针里。如果在,就延迟释放,或者放到一个待回收列表里。

// 伪代码:Hazard Pointer 的 pop 操作
pop():
    loop:
        old_head = head
        if old_head == NULL:
            return EMPTY
        // 把 old_head 注册为风险指针
        hazard_ptr = get_hazard_ptr_for_current_thread()
        hazard_ptr.set(old_head)
        // 再次检查 head 是否被修改(防止 ABA)
        if head != old_head:
            continue
        new_head = old_head.next
        if CAS(&head, old_head, new_head):
            // 成功 pop,解除风险指针
            hazard_ptr.clear()
            // 尝试回收 old_head
            retire_node(old_head)
            return old_head.value

我在项目中用过 Hazard Pointer,它的优点是实现相对简单,缺点是每个线程都要维护风险指针列表,而且回收时要做全局扫描,性能开销不小。

4.5.2 RCU(Read-Copy-Update)

RCU 是 Linux 内核里广泛使用的机制。它的核心思想是:读者不阻塞,写者等待所有读者退出临界区后再回收。

RCU 把操作分成三个阶段:

  1. 读端: 读者通过 rcu_dereference() 获取指针,然后直接访问。不需要锁,也不需要原子操作。
  2. 更新端: 写者创建一个新节点,用 rcu_assign_pointer() 原子地替换旧指针。这时候旧节点还在,读者可能还在用。
  3. 回收端: 写者调用 synchronize_rcu() 等待所有读者完成。等所有人都读完了,再安全地释放旧节点。
我的经验: RCU 在读多写少的场景下性能极好。比如路由表、配置信息这些几乎不更新的数据。我曾经在一个网络转发模块里用 RCU 替换读写锁,吞吐量直接翻了一倍。

但 RCU 也有局限:写者会被阻塞等待读者,而且实现起来比 Hazard Pointer 复杂得多。在用户态编程中,你需要自己模拟 RCU 的宽限期(grace period)检测。

4.6 知识结构总览

下面这张图总结了本章的核心知识点和它们之间的关系:

无锁编程核心知识体系 无锁栈 (Treiber Stack) 无锁队列 (Michael-Scott) 核心机制:CAS 原子操作 共同挑战:ABA 问题 + 内存回收 ABA 解决:标记指针 / 128位CAS 内存回收:Hazard Pointer / RCU Hazard Pointer vs RCU 对比 Hazard Pointer:读者开销大,写者延迟回收,适合通用场景 RCU:读者零开销,写者阻塞等待,适合读多写少场景

4.7 避坑指南与个人经验

最后,分享几个我踩过的坑:

  • 不要迷信无锁: 我曾经在一个锁冲突不严重的模块里强行用无锁队列,结果性能没提升,代码复杂度倒是翻了好几倍。无锁编程只有在锁竞争成为瓶颈时才值得用。
  • 内存序很重要: CAS 操作通常有 memory_order 参数。用错了内存序,轻则性能下降,重则逻辑错误。我建议在 x86 上先用 memory_order_seq_cst 保证正确性,优化时再放宽。
  • 测试要跑在真实硬件上: 无锁 bug 在模拟器或单核环境下很难复现。我曾经有一个 ABA 问题,在 8 核机器上跑了三天才稳定复现。所以,压力测试一定要上多核真实硬件。
  • 优先用现成的库: 除非你是为了学习,否则别自己造无锁轮子。C++11 的 std::atomic、Java 的 ConcurrentLinkedQueue、Linux 内核的 kfifo 都是经过千锤百炼的。
我曾经踩过的坑: 有一次我实现 Hazard Pointer 时,忘记在 pop 成功后重新检查 head 是否变化。结果线程 A 把节点 X 注册为风险指针后,线程 B 把 X pop 掉了,线程 A 还傻乎乎地以为 X 是安全的。最后查了三天才找到这个 bug。记住:注册风险指针后,一定要重新验证指针的有效性。

无锁编程是一门“看着简单,做起来难”的技术。它要求你对硬件、编译器、内存模型都有深刻的理解。但一旦掌握,它就是你手里最锋利的性能之刃。