第13章并发数据结构

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

13. 并发数据结构：无锁队列、无锁栈、读写分离的设计模式

并发编程里，数据结构这块儿，说白了就是「谁先碰、谁后碰、碰的时候别人能不能碰」的问题。锁当然能解决问题，但锁多了，性能就下来了。我早年做游戏服务器的时候，一个全局队列用互斥锁包着，压测一上来，CPU 全在等锁，吞吐量直接腰斩。嗯，从那以后，我对无锁数据结构就特别上心。

这一章，咱们聊聊三种常见的并发数据结构：无锁队列、无锁栈，以及读写分离的设计模式。它们不是银弹，但在合适的场景下，能让你把多核性能榨干。

13.1 无锁队列：生产者与消费者的桥梁

无锁队列，最经典的场景就是「一个生产者、一个消费者」（SPSC），或者「多个生产者、多个消费者」（MPMC）。你想想看，如果每次入队出队都要抢一把锁，那多核的优势就全浪费了。

我个人的习惯是，能用 SPSC 就别用 MPMC。因为 SPSC 可以用 ring buffer 实现，连 CAS 都不用，纯靠内存序保证可见性。来看一个简化版：

template<typename T, size_t Capacity>
class SPSCRingBuffer {
    static_assert((Capacity & (Capacity - 1)) == 0, "Capacity must be power of 2");
    std::array<T, Capacity> buffer_;
    std::atomic<size_t> head_{0};  // 消费者读的位置
    std::atomic<size_t> tail_{0};  // 生产者写的位置

public:
    bool push(const T& item) {
        size_t tail = tail_.load(std::memory_order_relaxed);
        size_t next = (tail + 1) & (Capacity - 1);
        if (next == head_.load(std::memory_order_acquire)) {
            return false; // 队列满
        }
        buffer_[tail] = item;
        tail_.store(next, std::memory_order_release);
        return true;
    }

    bool pop(T& item) {
        size_t head = head_.load(std::memory_order_relaxed);
        if (head == tail_.load(std::memory_order_acquire)) {
            return false; // 队列空
        }
        item = buffer_[head];
        head_.store((head + 1) & (Capacity - 1), std::memory_order_release);
        return true;
    }
};

这段代码里，memory_order_release 和 memory_order_acquire 是关键。它们保证了「写数据」一定在「更新 tail」之前完成，读数据一定在「读 head」之后开始。我曾经在项目里把 release 写成了 relaxed，结果消费者读到了半写的数据，排查了两天才找到原因……嗯，内存序这东西，真不能偷懒。

  核心要点： SPSC 无锁队列的核心是「环形缓冲区 + 内存序」。容量必须是 2 的幂，这样取模可以用位运算替代。

13.2 无锁栈：基于 CAS 的 LIFO 结构

无锁栈比队列简单一些，因为只需要操作一个头指针。但它的 ABA 问题很烦人。你想想看，线程 A 读到了 head 指针指向节点 X，然后被调度走了；线程 B 把 X 弹出了，又压入了另一个节点 Y，恰好 Y 的地址和 X 一样（被内存分配器复用了）。线程 A 醒来后，CAS 比较地址发现没变，就以为 head 还是原来的样子，结果把栈结构搞坏了。

解决 ABA 问题的经典做法是 带标记的指针（tagged pointer）。每次 CAS 时，不仅比较地址，还比较一个递增的版本号。C++ 标准库提供了 std::atomic<T*>，但没法直接存两个值。我建议用 std::atomic<uintptr_t> 自己编码：

template<typename T>
class LockFreeStack {
    struct Node {
        T data;
        Node* next;
    };
    std::atomic<uintptr_t> head_; // 低 48 位存指针，高 16 位存版本号

    static constexpr uintptr_t kTagMask = 0xFFFF000000000000ULL;
    static constexpr uintptr_t kPtrMask = ~kTagMask;

public:
    void push(T value) {
        Node* node = new Node{std::move(value), nullptr};
        uintptr_t old_head = head_.load(std::memory_order_relaxed);
        do {
            node->next = reinterpret_cast<Node*>(old_head & kPtrMask);
        } while (!head_.compare_exchange_weak(
            old_head,
            (reinterpret_cast<uintptr_t>(node) & kPtrMask) |
            ((old_head + 1) & kTagMask),
            std::memory_order_release,
            std::memory_order_relaxed));
    }

    bool pop(T& value) {
        uintptr_t old_head = head_.load(std::memory_order_relaxed);
        Node* ptr;
        do {
            ptr = reinterpret_cast<Node*>(old_head & kPtrMask);
            if (!ptr) return false;
            uintptr_t new_head =
                (reinterpret_cast<uintptr_t>(ptr->next) & kPtrMask) |
                ((old_head + 1) & kTagMask);
        } while (!head_.compare_exchange_weak(
            old_head, new_head,
            std::memory_order_acquire,
            std::memory_order_relaxed));
        value = std::move(ptr->data);
        delete ptr;
        return true;
    }
};

注意： 无锁栈的 ABA 问题不是「可能发生」，而是「一定会发生」。如果你不做版本标记，在高并发下迟早会踩坑。我曾经在一个 16 核的机器上跑测试，没加 tag 的版本跑了 3 分钟就崩了。

13.3 读写分离：读多写少的终极方案

读写分离，说白了就是「读的时候不阻塞写，写的时候不阻塞读」。最常见的实现是 RCU（Read-Copy-Update）。Linux 内核里大量用了这个模式。它的核心思想是：写操作先复制一份数据，在副本上修改，然后原子地替换指针；读操作直接读旧指针，等所有读操作都结束后，再释放旧数据。

用户态 C++ 实现 RCU 比较麻烦，因为需要知道「所有读操作是否都结束了」。我推荐一个轻量级的替代方案：双缓冲（Double Buffering）。

template<typename T>
class DoubleBuffer {
    std::array<T, 2> data_;
    std::atomic<int> active_{0}; // 0 或 1

public:
    const T& read() const {
        return data_[active_.load(std::memory_order_acquire)];
    }

    void write(const T& new_data) {
        int old_active = active_.load(std::memory_order_relaxed);
        int new_active = 1 - old_active;
        data_[new_active] = new_data;
        active_.store(new_active, std::memory_order_release);
    }
};

这个模式适合「读极多、写极少」的场景。比如配置表、路由表、黑白名单。我做过一个网络库，路由表用双缓冲，每秒几百万次查询，写操作一天才更新几次，性能非常稳。

小提示： 双缓冲的缺点是写操作会覆盖旧数据，如果写操作很频繁，读操作可能读到「半新半旧」的数据。这时候你需要考虑 RCU 或者读写锁。

13.4 如何选择？一张图说清楚

下面这张图总结了三种模式的适用场景和核心机制。你可以根据你的业务特点来选。

13.5 避坑指南

最后，分享几个我踩过的坑：

不要迷信无锁。 无锁结构在低竞争下可能比锁还慢（因为 CAS 循环会浪费 CPU）。我建议先上锁，压测发现瓶颈了再考虑无锁。
内存回收是难题。 无锁栈的 pop 里直接 delete 节点，如果另一个线程正持有该节点的指针，就悬空了。真正的无锁栈需要 延迟回收（如 hazard pointer 或 epoch-based reclamation）。
双缓冲不是线程安全的写。 如果两个线程同时调用 write，会互相覆盖。双缓冲只适合「单写多读」。
测试要上真实负载。 我曾经写了一个无锁队列，单线程测试全过，一上 8 线程压测就崩。原因是内存序写错了，导致一个线程的写入对另一个线程不可见。

嗯，并发数据结构这块儿，理论不难，难的是细节。你想想看，一个 memory_order 选错了，可能整个系统的正确性就没了。所以我的建议是：先理解原理，再动手实现，最后用工具（如 ThreadSanitizer）验证。这样踩坑的概率会小很多。

📚 课程目录

13. 并发数据结构：无锁队列、无锁栈、读写分离的设计模式

13.1 无锁队列：生产者与消费者的桥梁

13.2 无锁栈：基于 CAS 的 LIFO 结构

13.3 读写分离：读多写少的终极方案

13.4 如何选择？一张图说清楚

13.5 避坑指南