C++ 无锁队列：从 CAS 到内存序

背景

只要做过高并发服务、游戏引擎或者低延迟组件，迟早会碰到一个问题：锁太重了。

典型场景包括：

生产者线程持续推消息
消费者线程高频拉取任务
临界区很短，但锁竞争很激烈
延迟指标对尾部抖动非常敏感

这时候很多人第一反应是“上无锁队列”。

方向没错，但无锁代码最危险的地方在于：看起来能跑，不代表一定正确。

尤其在 C++ 里，只会用 compare_exchange_weak 还不够，真正决定正确性的往往是内存序。

无锁不等于没有同步

先澄清一个常见误区：

mutex 是同步
原子变量也是同步

无锁结构只是把同步方式从“阻塞锁”换成了“原子操作 + 内存可见性约束”。

也就是说，你不是不需要同步了，而是需要更精确地控制同步。

一个最简单的 SPSC 环形队列

先从单生产者、单消费者模型说起。这个模型更适合作为理解内存序的起点。

#include <atomic>
#include <array>
#include <cstddef>

template <typename T, std::size_t N>
class SpscQueue {
public:
    bool push(const T& value) {
        const auto tail = tail_.load(std::memory_order_relaxed);
        const auto next = (tail + 1) % N;

        if (next == head_.load(std::memory_order_acquire)) {
            return false;
        }

        buffer_[tail] = value;
        tail_.store(next, std::memory_order_release);
        return true;
    }

    bool pop(T& value) {
        const auto head = head_.load(std::memory_order_relaxed);

        if (head == tail_.load(std::memory_order_acquire)) {
            return false;
        }

        value = buffer_[head];
        head_.store((head + 1) % N, std::memory_order_release);
        return true;
    }

private:
    std::array<T, N> buffer_{};
    std::atomic<std::size_t> head_{0};
    std::atomic<std::size_t> tail_{0};
};

这个实现不复杂，但已经体现了两个关键点：

写入数据后，再发布 tail
读取 tail 时，要能看到生产者已经写好的数据

这就是 release / acquire 的配合。

release / acquire 到底在保证什么

很多人背过定义，但一写代码还是容易乱。

可以把它理解成一条很实用的规则：

release：我前面写的数据，现在可以安全对外发布了
acquire：我既然看到了这个发布结果，也必须能看到发布之前的写入

在上面的队列里：

生产者先把值写到 buffer_[tail]
然后 tail_.store(next, std::memory_order_release)
消费者读 tail_ 时使用 memory_order_acquire
这样消费者只要看见新的 tail，就一定能读到对应的 buffer 内容

如果这里全都写成 relaxed，在某些 CPU 或编译器重排下，消费者就可能先看到索引变化，却还没看到真正的数据写入。

这类 bug 最麻烦的地方在于：

本地开发机可能完全复现不了
压测时偶发
上线后高并发才暴露

为什么 MPMC 难很多

单生产者单消费者已经足够说明问题，但真实系统里常常是多生产者、多消费者。

这时复杂度会明显上升，因为你要同时处理：

多线程竞争同一个 head/tail
ABA 问题
节点回收时机
cache line false sharing

很多人会尝试链表式无锁队列，例如 Michael-Scott Queue。核心逻辑通常会长这样：

struct Node {
    int value;
    std::atomic<Node*> next{nullptr};
};

std::atomic<Node*> head;
std::atomic<Node*> tail;

void enqueue(Node* node) {
    node->next.store(nullptr, std::memory_order_relaxed);

    while (true) {
        Node* last = tail.load(std::memory_order_acquire);
        Node* next = last->next.load(std::memory_order_acquire);

        if (next == nullptr) {
            if (last->next.compare_exchange_weak(
                    next,
                    node,
                    std::memory_order_release,
                    std::memory_order_relaxed)) {
                tail.compare_exchange_weak(
                    last,
                    node,
                    std::memory_order_release,
                    std::memory_order_relaxed);
                return;
            }
        } else {
            tail.compare_exchange_weak(
                last,
                next,
                std::memory_order_release,
                std::memory_order_relaxed);
        }
    }
}

这里只是示意，不是生产可直接复制的完整实现。因为真正难的还不是 enqueue 本身，而是安全回收节点。

节点回收才是大坑

无锁链表或队列里，如果某个线程刚读到一个节点地址，另一个线程就把它删掉并复用内存，很容易出事。

这就是为什么很多成熟实现都会配套：

hazard pointers
epoch based reclamation
reference counting

如果你没有成熟的回收方案，只是把节点 delete 掉，那基本是在等线上事故。

所以工程上我通常这样判断：

如果是 SPSC，自己实现可控性还不错
如果是 MPMC，优先考虑成熟库
如果不是明显的锁竞争热点，就别为了“无锁”而无锁

很多人优化到原子操作这一层，却忽略了 cache line 竞争。

比如 head_ 和 tail_ 如果落在同一个 cache line，上下游线程频繁修改时，会不断触发 cache coherence 开销。

常见做法是手动对齐：

struct alignas(64) AlignedAtomicSize {
    std::atomic<std::size_t> value{0};
};

这类优化很朴素，但在高频路径上经常比“再抠一个 CAS”更有效。

什么时候值得上无锁队列

不是所有并发问题都值得上 lock-free。

我通常会先看这几个条件：

profile 明确显示锁竞争是热点
临界区非常短，锁开销占比明显
吞吐和尾延迟都很敏感
团队能维护这种复杂度

如果只是普通业务服务，mutex + condition_variable 很可能已经够用，而且可读性、可维护性更好。

总结

C++ 无锁编程最难的部分，不是 API，而是心智模型。

你必须真正理解：

原子操作在同步什么
内存序在保证什么
数据什么时候对其他线程可见
节点什么时候才能安全回收

这些问题搞不清楚，无锁结构往往只是把锁问题换成更难排查的一类并发 bug。

性能优化当然重要，但正确性始终排在前面。

并发代码里，最贵的从来不是多几条指令，而是一个难以稳定复现的竞态条件。

背景