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 同理。
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 落后了,就主动帮它更新。这种“互助”机制保证了队列不会卡死。
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 架构开始就普及了。
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 把操作分成三个阶段:
- 读端: 读者通过 rcu_dereference() 获取指针,然后直接访问。不需要锁,也不需要原子操作。
- 更新端: 写者创建一个新节点,用 rcu_assign_pointer() 原子地替换旧指针。这时候旧节点还在,读者可能还在用。
- 回收端: 写者调用 synchronize_rcu() 等待所有读者完成。等所有人都读完了,再安全地释放旧节点。
但 RCU 也有局限:写者会被阻塞等待读者,而且实现起来比 Hazard Pointer 复杂得多。在用户态编程中,你需要自己模拟 RCU 的宽限期(grace period)检测。
4.6 知识结构总览
下面这张图总结了本章的核心知识点和它们之间的关系:
4.7 避坑指南与个人经验
最后,分享几个我踩过的坑:
- 不要迷信无锁: 我曾经在一个锁冲突不严重的模块里强行用无锁队列,结果性能没提升,代码复杂度倒是翻了好几倍。无锁编程只有在锁竞争成为瓶颈时才值得用。
- 内存序很重要: CAS 操作通常有 memory_order 参数。用错了内存序,轻则性能下降,重则逻辑错误。我建议在 x86 上先用 memory_order_seq_cst 保证正确性,优化时再放宽。
- 测试要跑在真实硬件上: 无锁 bug 在模拟器或单核环境下很难复现。我曾经有一个 ABA 问题,在 8 核机器上跑了三天才稳定复现。所以,压力测试一定要上多核真实硬件。
- 优先用现成的库: 除非你是为了学习,否则别自己造无锁轮子。C++11 的
std::atomic、Java 的ConcurrentLinkedQueue、Linux 内核的kfifo都是经过千锤百炼的。
无锁编程是一门“看着简单,做起来难”的技术。它要求你对硬件、编译器、内存模型都有深刻的理解。但一旦掌握,它就是你手里最锋利的性能之刃。