无锁编程:CAS操作、ABA问题、无锁数据结构的实现

说到并发编程,大家第一反应肯定是加锁。互斥锁、读写锁、自旋锁……锁确实好用,但锁也有它的代价。上下文切换、线程阻塞、死锁风险,这些都是锁带来的麻烦。

那有没有不用锁也能保证线程安全的方法?有,这就是无锁编程。说白了,就是利用硬件提供的原子操作,直接在CPU指令层面搞定同步。今天我们就来聊聊无锁编程的核心——CAS操作,以及它带来的ABA问题,最后看看怎么实现一个无锁数据结构。

核心观点:无锁编程不是真的没有锁,而是把锁的粒度降到了CPU指令级别。它避免了操作系统级别的线程阻塞,所以在高并发场景下性能往往更好。

一、CAS操作:无锁编程的基石

CAS,全称是Compare-And-Swap,比较并交换。它的工作逻辑很简单:

  1. 读取内存中的值V
  2. 判断V是否等于预期值A
  3. 如果相等,把V更新为新值B
  4. 如果不相等,什么都不做

整个过程是原子的,由CPU硬件保证。我当年第一次看到这个操作时,心里想的是:这不就是个if语句吗?怎么就成原子操作了?后来才明白,关键在于CPU提供了CMPXCHG这样的指令,一条指令完成比较和交换,中间不会被中断。

在C语言中,C11标准提供了<stdatomic.h>头文件,我们可以这样使用CAS:

#include <stdatomic.h>

atomic_int counter = 0;

void increment() {
    int expected = atomic_load(&counter);
    int desired = expected + 1;
    
    // 尝试CAS,如果失败就重试
    while (!atomic_compare_exchange_weak(&counter, &expected, desired)) {
        desired = expected + 1;
    }
}

这里有个细节:atomic_compare_exchange_weak可能会因为硬件原因假失败,所以需要放在循环里重试。而atomic_compare_exchange_strong则保证要么成功要么失败,不会假失败。我个人习惯在循环里用weak版本,性能更好一些。

小技巧:CAS操作通常配合循环使用,这就是所谓的"CAS循环"或"自旋CAS"。如果竞争不激烈,循环一两次就成功了;如果竞争激烈,可能会循环很多次,这时候性能就不如锁了。

二、ABA问题:一个隐蔽的陷阱

CAS操作看起来完美,但有一个经典问题——ABA问题。我在项目中就踩过这个坑,当时排查了好久才找到原因。

什么是ABA问题?假设有一个共享变量V,初始值是A:

  1. 线程1读取V,得到A
  2. 线程2把V从A改成B,再改回A
  3. 线程1执行CAS,发现V还是A,CAS成功

问题出在哪?线程1以为V没变过,但实际上V经历了A→B→A的变化。在某些场景下,这种"中间变化"会导致严重错误。

举个我遇到的例子:一个无锁栈,节点通过指针链接。线程1要弹出栈顶节点A,它读取了A的指针。但此时线程2插进来,把A弹出,又把A重新压入栈。线程1继续执行CAS,发现栈顶指针还是A,于是成功弹出。但A的内部状态已经变了,程序就乱了。

注意:ABA问题不是CAS本身的缺陷,而是使用CAS时需要考虑的语义问题。如果数据结构的正确性依赖于"值没有发生过变化",那么ABA问题就可能出现。

解决ABA问题,常见的方法有两种:

方法 原理 适用场景
标记指针(Tagged Pointer) 在指针中嵌入一个版本号或标记位,每次修改都递增标记 32位或64位指针,有冗余位可用
双CAS(DCAS) 同时CAS两个独立的内存位置 硬件支持DCAS的场景
垃圾回收(RCU/GC) 延迟回收内存,确保不会重用 有GC的语言或使用RCU机制

在C语言中,最常用的是标记指针。比如在x86-64架构下,用户空间指针的高16位通常为0,我们可以利用这些位来存储版本号:

// 假设指针只有低48位有效,高16位用来存版本号
typedef struct {
    uintptr_t ptr;  // 低48位是指针,高16位是版本号
} tagged_ptr_t;

tagged_ptr_t atomic_load_tagged(atomic_uintptr_t *p) {
    uintptr_t value = atomic_load(p);
    return (tagged_ptr_t){.ptr = value};
}

bool atomic_cas_tagged(atomic_uintptr_t *p, tagged_ptr_t *expected, tagged_ptr_t desired) {
    uintptr_t old = expected->ptr;
    uintptr_t new_val = desired.ptr;
    return atomic_compare_exchange_strong(p, &old, new_val);
}

嗯,这里要注意:标记指针虽然好用,但需要确保你的平台确实有空闲位可用。我记得有一次在32位系统上移植代码,发现指针全占满了,根本没法塞版本号,最后只能改用其他方案。

三、无锁数据结构的实现

理论说完了,我们来点实际的。实现一个无锁栈,这是无锁数据结构中最经典的例子。

无锁栈的核心思想:用CAS操作来更新栈顶指针。每次push或pop时,先读取栈顶指针,然后尝试用CAS更新它。如果CAS失败,说明有其他线程修改了栈顶,就重试。

#include <stdatomic.h>
#include <stdlib.h>

typedef struct Node {
    int value;
    struct Node *next;
} Node;

typedef struct {
    atomic_uintptr_t top;  // 栈顶指针,带标记
} LockFreeStack;

void push(LockFreeStack *stack, int value) {
    Node *new_node = (Node *)malloc(sizeof(Node));
    new_node->value = value;
    
    uintptr_t old_top = atomic_load(&stack->top);
    uintptr_t new_top;
    
    do {
        // 新节点的next指向当前栈顶
        new_node->next = (Node *)(old_top & 0x0000FFFFFFFFFFFF);
        // 构造新的栈顶指针(低48位是新节点地址,高16位版本号+1)
        uintptr_t version = (old_top >> 48) + 1;
        new_top = ((uintptr_t)new_node & 0x0000FFFFFFFFFFFF) | (version << 48);
    } while (!atomic_compare_exchange_weak(&stack->top, &old_top, new_top));
}

int pop(LockFreeStack *stack, int *value) {
    uintptr_t old_top = atomic_load(&stack->top);
    uintptr_t new_top;
    Node *node;
    
    do {
        node = (Node *)(old_top & 0x0000FFFFFFFFFFFF);
        if (node == NULL) {
            return -1;  // 栈空
        }
        
        // 新栈顶指向原栈顶的下一个节点
        uintptr_t next_ptr = (uintptr_t)node->next;
        uintptr_t version = (old_top >> 48) + 1;
        new_top = (next_ptr & 0x0000FFFFFFFFFFFF) | (version << 48);
    } while (!atomic_compare_exchange_weak(&stack->top, &old_top, new_top));
    
    *value = node->value;
    free(node);
    return 0;
}

这段代码里,我用标记指针解决了ABA问题。每次push或pop时,版本号都会递增。即使指针值相同,版本号不同,CAS也会失败,从而避免了ABA问题。

关键点:无锁栈的push和pop操作都是"乐观的"——先假设没有竞争,直接操作;如果发现竞争(CAS失败),就重试。这种乐观策略在低竞争场景下性能极好。

你可能会问:为什么不用锁?锁不是更简单吗?确实,如果并发度不高,用锁完全没问题。但如果你在做高并发服务器、实时系统或者需要避免死锁的场景,无锁数据结构就很有价值了。

我曾经在一个消息中间件项目里,用无锁队列替换了加锁队列,吞吐量提升了将近3倍。当然,代价是代码复杂度增加了不少,调试也更困难。所以我的建议是:

  • 如果并发度不高(几十个线程以内),用锁就够了
  • 如果追求极致性能,或者需要避免死锁,再考虑无锁
  • 无锁代码一定要经过严格的测试,特别是压力测试

四、知识体系总览

下面这张图总结了无锁编程的核心知识结构,你可以对照着梳理思路:

无锁编程知识体系 CAS操作(核心) 工作原理 比较并交换,硬件原子指令 C11 stdatomic.h 支持 ABA问题 值被修改后又恢复原样 解决方案:标记指针/DCAS/RCU 无锁数据结构 无锁栈(LIFO) 无锁队列(FIFO) 核心思想 乐观并发控制:假设无竞争,CAS失败则重试 避免操作系统级阻塞,适合高并发低竞争场景

无锁编程是一门"看着简单,用起来难"的技术。CAS操作本身只有几条指令,但要把它们组合成正确的数据结构,需要考虑很多边界情况。我建议你从无锁栈开始练手,然后尝试实现无锁队列,最后再挑战更复杂的结构如无锁哈希表。

记住一句话:无锁编程不是银弹。它适合特定场景,但不要为了炫技而滥用。在实际项目中,我通常先用锁实现一个版本,性能测试确认瓶颈在锁上之后,再考虑无锁优化。

避坑指南:我曾经在无锁代码中忘记处理内存序(memory order),导致在ARM架构上出现诡异的数据不一致问题。后来老老实实把每个原子操作的内存序都标注清楚,问题就解决了。所以,不要依赖默认的内存序,显式指定memory_order_acquirememory_order_releasememory_order_seq_cst

好了,关于无锁编程的核心内容就这些。CAS操作是基础,ABA问题是陷阱,标记指针是常用解法,无锁栈是入门实践。把这些搞明白,你就掌握了无锁编程的"三板斧"。

公众号:蓝海资料掘金营,微信deep3321