第4章无锁数据结构

第4章：无锁数据结构：CAS原语、内存序、无锁队列与无锁栈

各位同学，欢迎来到第四章。这一章我们聊点硬核的——无锁数据结构。

说实话，我刚开始做量化交易系统那会儿，对锁的态度就是“能用锁解决的事，都不是事”。直到有一次，我在一个高频行情处理模块里，用了一个粗粒度的互斥锁保护一个共享队列。结果呢？行情峰值一来，延迟直接飙到微秒级，策略信号全乱套了。那次之后，我彻底明白了：在低延迟场景下，锁就是性能杀手。

所以，无锁数据结构不是炫技，是刚需。

4.1 CAS原语：无锁的基石

无锁编程的核心，说白了就是一条CPU指令——CAS（Compare-And-Swap）。

它的语义很简单：比较一个内存地址的值是否等于期望值，如果相等，就把它更新为新值。整个过程是原子的，不会被线程调度打断。

我习惯用C++11提供的std::atomic来操作CAS，而不是直接写内联汇编。可读性更好，编译器也会帮你做优化。

#include <atomic>

std::atomic<int> counter{0};

// CAS 操作：如果 counter 等于 0，就把它改成 1
int expected = 0;
bool success = counter.compare_exchange_strong(expected, 1);
// 如果 success 为 true，说明 CAS 成功，counter 现在是 1
// 如果 success 为 false，说明 counter 不是 0，expected 会被更新为当前值

这里有个坑，我踩过。 compare_exchange_strong 和 compare_exchange_weak 有什么区别？

强版本保证要么成功要么失败，不会出现“假失败”。弱版本在某些架构上（比如ARM）可能出现假失败，但性能更好。我个人的建议是：在循环里用弱版本，在非循环场景用强版本。

避坑指南： 我曾经在ARM服务器上写了一个无锁栈，用了强CAS，结果性能死活上不去。换成弱CAS后，吞吐量提升了30%。记住，弱CAS不是bug，是特性。

4.2 内存序（memory_order）：看不见的约束

CAS只是工具，真正让无锁数据结构跑对的是内存序。你想想看，多核CPU为了性能，会乱序执行指令，还会把写操作缓存到自己的L1 cache里。如果没有内存序约束，一个线程写了一个值，另一个线程可能根本看不到。

C++11定义了六种内存序，我挑三个最常用的说：

内存序	含义	性能开销	使用场景
`memory_order_relaxed`	只保证原子性，不保证任何顺序	几乎为零	计数器、统计量
`memory_order_acquire`	保证后续读操作不会被重排到前面	中等	读取锁状态
`memory_order_release`	保证前面的写操作不会被重排到后面	中等	写入锁状态
`memory_order_seq_cst`	全局顺序一致性，最严格	最高	默认值，但慎用

我刚开始写无锁代码时，所有原子操作都用了memory_order_seq_cst。结果呢？性能比用锁还差。后来我学乖了，能用relaxed就用relaxed，需要同步时才用acquire/release。

我的经验： 如果你不确定该用哪个内存序，先用seq_cst把逻辑写对，然后逐步放松。每次放松后跑一遍压力测试，确保不出问题。

4.3 无锁队列：MPSC与SPSC

在交易引擎里，队列是最常用的数据结构。行情数据从网卡到策略引擎，中间要经过好几层队列。如果队列有锁，延迟就上去了。

我重点讲两种：SPSC（单生产者单消费者）和MPSC（多生产者单消费者）。

4.3.1 SPSC队列：最简单的无锁队列

SPSC队列只有一个生产者和一个消费者，所以不需要复杂的同步。只需要一个读指针和一个写指针，用relaxed内存序就够了。

template<typename T, size_t Capacity>
class SPSCQueue {
    std::array<T, Capacity> buffer_;
    std::atomic<size_t> read_pos_{0};
    std::atomic<size_t> write_pos_{0};

public:
    bool push(const T& item) {
        size_t write = write_pos_.load(std::memory_order_relaxed);
        size_t next_write = (write + 1) % Capacity;
        if (next_write == read_pos_.load(std::memory_order_acquire)) {
            return false; // 队列满了
        }
        buffer_[write] = item;
        write_pos_.store(next_write, std::memory_order_release);
        return true;
    }

    bool pop(T& item) {
        size_t read = read_pos_.load(std::memory_order_relaxed);
        if (read == write_pos_.load(std::memory_order_acquire)) {
            return false; // 队列空了
        }
        item = buffer_[read];
        read_pos_.store((read + 1) % Capacity, std::memory_order_release);
        return true;
    }
};

你看，代码很简洁。但有个细节要注意：push里读read_pos_用了acquire，pop里读write_pos_也用了acquire。这是为了确保能看到对方的最新写入。

  核心要点： SPSC队列的性能极高，在Intel CPU上，一次push/pop的开销大约在10-20纳秒。如果你的场景是单生产者单消费者，优先用SPSC。

4.3.2 MPSC队列：多生产者怎么办？

现实场景中，往往有多个线程往同一个队列里写数据。比如多个行情网关同时写入一个队列。这时候SPSC就不够用了。

MPSC队列的核心思路是：多个生产者通过CAS竞争写权限，消费者单线程读。

我常用的实现是：生产者先CAS更新写指针，如果成功，就独占写入槽位。如果失败，就自旋重试。

bool push(const T& item) {
    size_t write = write_pos_.load(std::memory_order_relaxed);
    size_t next_write;
    do {
        next_write = (write + 1) % Capacity;
        if (next_write == read_pos_.load(std::memory_order_acquire)) {
            return false; // 满了
        }
    } while (!write_pos_.compare_exchange_weak(write, next_write,
                std::memory_order_release, std::memory_order_relaxed));
    buffer_[write] = item;
    return true;
}

这里用了compare_exchange_weak在循环里，因为多个生产者可能同时竞争，CAS失败是常态。我建议把自旋次数限制一下，比如最多重试100次，否则就返回失败。防止某个生产者一直抢不到。

注意： 我曾经在生产环境遇到过一个问题：某个生产者线程被操作系统调度走了，导致其他生产者一直CAS失败，队列吞吐量骤降。解决方案是加一个超时机制，或者用更高级的“无等待”算法。

4.4 无锁栈：LIFO的原子实现

栈在交易引擎里用得少一些，但偶尔也会用到，比如内存池的回收。无锁栈的实现比队列简单，核心是一个CAS操作头指针。

template<typename T>
class LockFreeStack {
    struct Node {
        T data;
        Node* next;
    };
    std::atomic<Node*> head_{nullptr};

public:
    void push(T value) {
        Node* new_node = new Node{value, nullptr};
        new_node->next = head_.load(std::memory_order_relaxed);
        while (!head_.compare_exchange_weak(new_node->next, new_node,
                    std::memory_order_release, std::memory_order_relaxed)) {
            // CAS失败时，new_node->next已经被更新为最新的head
        }
    }

    bool pop(T& value) {
        Node* old_head = head_.load(std::memory_order_relaxed);
        while (old_head && !head_.compare_exchange_weak(old_head, old_head->next,
                    std::memory_order_acquire, std::memory_order_relaxed)) {
            // 重试
        }
        if (!old_head) return false;
        value = old_head->data;
        delete old_head; // 注意：这里可能有ABA问题
        return true;
    }
};

嗯，这里有个经典问题——ABA问题。假设线程A读取head为节点X，然后被挂起。线程B把X弹出，又压入一个节点Y，但Y的地址恰好和X一样（被内存回收后重新分配）。线程A醒来后，CAS发现head还是X（实际上是Y），就认为没有变化，结果把Y错误地弹出了。

怎么解决？我常用的方法是给指针加一个版本号，用双字CAS（DCAS）或者用std::atomic<uintptr_t>把指针和版本号打包在一起。

我的建议： 在交易引擎里，如果栈的节点数量不大，可以用内存池预分配，避免内存回收导致的ABA问题。这样既解决了ABA，又避免了动态内存分配的开销。

4.5 知识体系总览

下面这张图是我自己总结的无锁数据结构知识体系，你可以把它当作一个思维导图来用。

这张图把本章的核心知识点串起来了。从底层的CAS原语，到内存序的约束，再到上层的队列和栈。你写代码的时候，可以对照这张图，看看自己用到了哪些技术。

好了，这一章的内容就到这里。无锁数据结构是低延迟交易引擎的基石，但也是容易出错的地方。我建议你从SPSC队列开始练手，写一个能跑通的版本，然后加上压力测试，看看性能到底能到多少纳秒。只有亲手写过，才能真正理解这些概念。

📚 课程目录