3. 基于环形缓冲区的逐笔消息零拷贝传递实现

为什么要把队列换成环形缓冲区

解析线程与策略线程之间常见的传递方式是 std::queue<Tick> 加互斥锁，或者 std::mutex + condition_variable。这条路径上有三处固定开销：入队/出队各一次加锁、每次 push 可能触发 std::deque 的节点分配、以及消费者被唤醒时的调度延迟。逐笔行情的特点是单生产者、单消费者、消息短小、速率高，这正好适合用无锁环形缓冲区（SPSC ring buffer）替代。

本案例的“零拷贝”指的不是网络层的零拷贝，而是消息在解析线程与策略线程之间不经过中间容器的复制与堆分配：解析线程直接在环形缓冲区的槽位里原地构造 Tick，策略线程直接读取该槽位，槽位内存由缓冲区一次性预分配并循环复用。这样每条消息省掉一次 new、一次 memcpy 和两次锁操作。

环形缓冲区的结构与内存序约束

SPSC 环形缓冲区只需要两个原子游标：生产者写 write_idx，消费者写 read_idx，容量取 2 的幂以便用位掩码取模。核心约束是：生产者必须先写入槽位数据，再发布 write_idx；消费者必须先读到新的 write_idx，再读槽位数据。因此生产者用 release 语义发布游标，消费者用 acquire 语义读取游标，否则编译器或 CPU 可能把槽位写入重排到游标更新之后，导致消费者读到半条消息。

可运行的 SPSC 环形缓冲区实现

下面是一个完整可编译的最小实现，包含缓冲区本体、生产者/消费者接口，以及一个用于验证吞吐与延迟的测试主程序。容量固定为 2 的幂，Tick 为定长结构体，槽位数组在构造时一次性分配。

#include <atomic>
#include <cstdint>
#include <cstdio>
#include <chrono>
#include <thread>
#include <vector>
#include <algorithm>

struct Tick {
    uint64_t seq;        // 交易所序列号
    uint64_t ts_ns;      // 本地接收时间戳
    uint32_t price;      // 价格（定点）
    uint32_t qty;        // 数量
    uint8_t  side;       // 0=买 1=卖
};

template <typename T, size_t Capacity>
class SpscRing {
    static_assert((Capacity & (Capacity - 1)) == 0, "Capacity 必须是 2 的幂");
public:
    SpscRing() : slots_(Capacity) {}

    // 生产者：返回可原地构造的槽位指针，失败返回 nullptr（满）
    T* try_acquire_slot() {
        const size_t w = write_idx_.load(std::memory_order_relaxed);
        const size_t r = read_idx_.load(std::memory_order_acquire);
        if (w - r == Capacity) return nullptr;      // 满
        return &slots_[w & (Capacity - 1)];
    }

    // 生产者：槽位构造完成后发布
    void publish() {
        const size_t w = write_idx_.load(std::memory_order_relaxed);
        write_idx_.store(w + 1, std::memory_order_release);
    }

    // 消费者：取可读槽位，失败返回 nullptr（空）
    const T* try_peek() const {
        const size_t r = read_idx_.load(std::memory_order_relaxed);
        const size_t w = write_idx_.load(std::memory_order_acquire);
        if (r == w) return nullptr;                 // 空
        return &slots_[r & (Capacity - 1)];
    }

    // 消费者：消费完成后释放槽位
    void release() {
        const size_t r = read_idx_.load(std::memory_order_relaxed);
        read_idx_.store(r + 1, std::memory_order_release);
    }

private:
    std::vector<T> slots_;
    alignas(64) std::atomic<size_t> write_idx_{0};
    alignas(64) std::atomic<size_t> read_idx_{0};
};

注意 write_idx_ 与 read_idx_ 各自用 alignas(64) 对齐到独立缓存行，避免生产者与消费者互相把对方的游标所在缓存行置为失效（伪共享）。这是 SPSC 缓冲区在高吞吐下能否真正跑出低延迟的关键细节之一。

吞吐与延迟验证程序

测试程序让生产者连续写入 N 条消息，消费者逐条读取并记录“从生产者发布到消费者读到”的延迟。为了让延迟测量有意义，生产者在 Tick.ts_ns 中写入发布前的 steady_clock 时间戳，消费者读到后立即取当前时间相减。这是教学示例，用于对比两种传递方式的相对差异，不代表真实交易所链路的绝对延迟。

int main() {
    constexpr size_t N = 5'000'000;
    SpscRing<Tick, 4096> ring;
    std::atomic<bool> done{false};
    std::vector<uint64_t> latencies;
    latencies.reserve(N);

    auto t0 = std::chrono::steady_clock::now();

    std::thread consumer([&] {
        size_t got = 0;
        while (got < N) {
            const Tick* t = ring.try_peek();
            if (!t) { std::this_thread::yield(); continue; }
            uint64_t now = std::chrono::duration_cast<std::chrono::nanoseconds>(
                std::chrono::steady_clock::now().time_since_epoch()).count();
            latencies.push_back(now - t->ts_ns);
            ring.release();
            ++got;
        }
        done.store(true, std::memory_order_release);
    });

    std::thread producer([&] {
        for (size_t i = 0; i < N; ++i) {
            Tick* slot;
            while (!(slot = ring.try_acquire_slot())) std::this_thread::yield();
            slot->seq = i;
            slot->ts_ns = std::chrono::duration_cast<std::chrono::nanoseconds>(
                std::chrono::steady_clock::now().time_since_epoch()).count();
            slot->price = 10000 + (i % 100);
            slot->qty = 100;
            slot->side = i & 1;
            ring.publish();
        }
    });

    producer.join();
    consumer.join();
    auto t1 = std::chrono::steady_clock::now();

    double secs = std::chrono::duration<double>(t1 - t0).count();
    std::sort(latencies.begin(), latencies.end());
    auto pct = [&](double p) { return latencies[(size_t)(p * (latencies.size() - 1))]; };

    printf("消息数: %zu\n", N);
    printf("吞吐: %.2f M msg/s\n", N / secs / 1e6);
    printf("延迟 P50: %lu ns\n", pct(0.50));
    printf("延迟 P99: %lu ns\n", pct(0.99));
    printf("延迟 P999: %lu ns\n", pct(0.999));
    return 0;
}

如何判读结果并确认零拷贝路径生效

验证分两层。第一层是功能正确性：消费者收到的 seq 必须严格连续递增，若出现跳号或重复，说明游标发布/读取的内存序写错了，或容量掩码用错。第二层是性能改善：在同一台机器、同一编译选项下，分别跑 mutex+queue 版本和 ring buffer 版本，对比吞吐与 P99 延迟。

落地检查清单

• 容量是否为 2 的幂，掩码是否为 Capacity - 1。
• 生产者发布用 release，消费者读取 write_idx 用 acquire，两者不可省。
• 两个游标是否分处不同缓存行（alignas(64)）。
• 槽位数组是否在构造时一次性分配，运行期是否零 new。
• 消费者是否直接读槽位而非拷贝到临时对象后再处理。
• 是否记录并监控缓冲区满/空自旋次数，作为容量是否足够的依据。