第12章并发数据结构

课程目录第12章并发数据结构

并发数据结构：无锁队列的原理

说到并发编程，锁是个绕不开的话题。但锁用多了，性能就下来了。我记得几年前在一个高吞吐量的消息中间件项目里，锁竞争直接成了瓶颈。那时候我就在想——有没有办法不用锁，也能保证数据安全？

答案就是无锁数据结构。今天咱们重点聊聊无锁队列。

为什么需要无锁队列？

传统的互斥锁队列，说白了就是「一次只让一个人干活」。线程A在push，线程B就得等着。这在高并发场景下，上下文切换的开销非常可观。

我遇到过最极端的情况：一个8核的机器，用锁队列跑压测，CPU利用率只有30%——剩下的全耗在锁等待和线程切换上了。换成无锁队列后，CPU利用率直接飙到85%，吞吐量翻了将近3倍。

无锁队列的核心思想是：让多个线程同时读写，但通过原子操作保证数据的一致性。谁先完成操作，谁的数据就生效。冲突了？重试一次就好。

CAS操作与原子操作

无锁队列的基石，是CAS（Compare-And-Swap）操作。

CAS是个硬件级别的原子指令。它的逻辑很简单：

// 伪代码：CAS(&addr, expected, new_value)
// 如果 *addr == expected，就把 *addr 设为 new_value，返回 true
// 否则返回 false
bool CAS(int* addr, int expected, int new_value) {
    if (*addr == expected) {
        *addr = new_value;
        return true;
    }
    return false;
}

注意，这个「比较+交换」是一气呵成的，中间不会被其他线程打断。这就是原子性。

我个人的习惯是，把CAS想象成「乐观锁」——先假设没人跟我抢，直接动手改。如果发现别人改过了（值不等于expected），那就重来。

核心要点：CAS操作是硬件保证的原子操作，它解决了「读-改-写」这个复合操作的并发安全问题。

内存屏障的概念

嗯，这里有个坑。CAS虽然保证了原子性，但它管不了内存可见性。

现代CPU为了性能，会乱序执行指令。线程A写了一个值，线程B可能看不到最新的结果——因为写操作还在CPU的store buffer里没刷到主存。

这就是内存屏障（Memory Barrier）要解决的问题。

内存屏障分几种：

类型	作用
读屏障（Load Barrier）	保证屏障之后的读操作，能看到屏障之前所有写操作的结果
写屏障（Store Barrier）	强制将store buffer中的数据刷到主存
全屏障（Full Barrier）	同时保证读写操作的顺序和可见性

我曾经在一个无锁队列的bug排查中，花了整整两天才定位到问题——就是少了一个写屏障。线程A把数据写入了队列，但线程B读到的全是旧数据。加上屏障后，问题立刻消失。

避坑指南：不要以为CAS操作就万事大吉了。CAS只保证「比较并交换」这个动作的原子性，不保证内存可见性。在多核CPU上，必须配合内存屏障使用。

C11原子操作库

C11标准引入了 <stdatomic.h>，终于让C语言有了跨平台的原子操作支持。以前写无锁代码，得用GCC的 __sync_* 或者Windows的 Interlocked*，移植性很差。

C11原子操作库的核心类型是 atomic_xxx：

#include <stdatomic.h>

atomic_int counter = 0;
atomic_store(&counter, 10);          // 原子写
int val = atomic_load(&counter);     // 原子读
int old = atomic_fetch_add(&counter, 1); // 原子加，返回旧值

最常用的是 atomic_compare_exchange_strong，也就是CAS的C11版本：

atomic_int head = 0;
int expected = 0;
int desired = 1;

// 如果 head == expected，就把 head 设为 desired
if (atomic_compare_exchange_strong(&head, &expected, desired)) {
    // CAS成功
} else {
    // CAS失败，expected会被更新为head的当前值
}

这里有个细节：atomic_compare_exchange_strong 的第二个参数是指针，失败后会自动更新为最新值。这样你就不用再手动读一次了。

C11还提供了内存序参数，用来控制内存屏障的强度：

memory_order_relaxed：最宽松，只保证原子性，不保证顺序
memory_order_acquire：读操作后的代码不能重排到读之前
memory_order_release：写操作前的代码不能重排到写之后
memory_order_seq_cst：最严格，全局顺序一致

我个人建议，刚开始用C11原子操作时，先全部用 memory_order_seq_cst。虽然性能差一点，但不容易出错。等跑通了，再根据场景优化内存序。

一个简单的无锁队列实现

说了这么多，咱们看个实际例子。这是一个基于链表的无锁队列（多生产者/多消费者）：

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

typedef struct Node {
    void* data;
    struct Node* next;
} Node;

typedef struct {
    atomic_int count;      // 队列元素个数
    atomic_uintptr_t head; // 头指针（用于出队）
    atomic_uintptr_t tail; // 尾指针（用于入队）
} LockFreeQueue;

// 初始化队列
void queue_init(LockFreeQueue* q) {
    Node* dummy = (Node*)malloc(sizeof(Node));
    dummy->next = NULL;
    atomic_store(&q->head, (uintptr_t)dummy);
    atomic_store(&q->tail, (uintptr_t)dummy);
    atomic_store(&q->count, 0);
}

// 入队
void queue_push(LockFreeQueue* q, void* data) {
    Node* node = (Node*)malloc(sizeof(Node));
    node->data = data;
    node->next = NULL;

    Node* tail;
    Node* next;
    while (1) {
        tail = (Node*)atomic_load(&q->tail);
        next = (Node*)atomic_load(&tail->next);
        if (tail == (Node*)atomic_load(&q->tail)) {
            if (next == NULL) {
                // 尝试把新节点链接到尾部
                if (atomic_compare_exchange_strong(
                        &tail->next, &next, node)) {
                    break; // 链接成功
                }
            } else {
                // 尾部落后了，帮它前移
                atomic_compare_exchange_strong(
                    &q->tail, &tail, next);
            }
        }
    }
    // 移动tail指针
    atomic_compare_exchange_strong(
        &q->tail, &tail, node);
    atomic_fetch_add(&q->count, 1);
}

// 出队
void* queue_pop(LockFreeQueue* q) {
    Node* head;
    Node* tail;
    Node* next;
    void* data;

    while (1) {
        head = (Node*)atomic_load(&q->head);
        tail = (Node*)atomic_load(&q->tail);
        next = (Node*)atomic_load(&head->next);

        if (head == (Node*)atomic_load(&q->head)) {
            if (head == tail) {
                if (next == NULL) {
                    return NULL; // 队列为空
                }
                // tail落后了，帮它前移
                atomic_compare_exchange_strong(
                    &q->tail, &tail, next);
            } else {
                data = next->data;
                // 尝试移动head指针
                if (atomic_compare_exchange_strong(
                        &q->head, &head, next)) {
                    free(head); // 释放旧的dummy节点
                    atomic_fetch_sub(&q->count, 1);
                    return data;
                }
            }
        }
    }
}

这个实现里有个巧妙的设计——dummy节点。它让队列永远不为空（至少有一个dummy节点），这样入队和出队操作就不会互相干扰。

小技巧：无锁队列的调试非常困难。我建议先用单线程跑通逻辑，再用两个线程做压力测试。出问题的话，先检查内存序，再检查CAS的条件判断。

知识体系总览

下面这张图总结了无锁队列的核心知识脉络：

说白了，无锁队列就是「用硬件指令代替锁，用重试代替阻塞」。它不适合所有场景——如果冲突率太高，CAS重试的开销反而比锁更大。但如果你能控制好数据分布和线程数量，无锁队列带来的性能提升是立竿见影的。

我当年第一次把无锁队列部署到生产环境时，心里其实挺没底的。但跑了一周后，监控数据告诉我：延迟降低了60%，吞吐量翻倍。嗯，从那以后，我就成了无锁编程的忠实拥趸。

📚 课程目录

并发数据结构：无锁队列的原理

为什么需要无锁队列？

CAS操作与原子操作

内存屏障的概念

C11原子操作库

一个简单的无锁队列实现

知识体系总览