第17章并发数据结构

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

并发数据结构：无锁队列、无锁栈、读写分离、RCU

并发数据结构这块，说实话是很多C++后端工程师的噩梦。我早年接手过一个高并发网关项目，锁竞争导致CPU利用率上不去，吞吐量卡在瓶颈上。后来把核心队列换成无锁实现，性能直接翻倍。今天我就把这几类常用的并发数据结构掰开揉碎了讲清楚。

为什么需要无锁数据结构？

传统的互斥锁（mutex）在高并发场景下有个致命问题：线程阻塞和上下文切换。你想想看，一个线程持有锁，其他线程只能干等着。如果临界区代码很短，锁的开销甚至比实际工作还大。

无锁数据结构利用原子操作（CAS、FAA等）来保证线程安全，避免了操作系统介入的锁机制。说白了，就是让多个线程同时操作数据结构，但通过硬件级别的原子指令来保证一致性。

核心思想：无锁 ≠ 没有同步，而是将同步粒度降到CPU指令级别。

无锁队列（Lock-Free Queue）

无锁队列是最常见的无锁数据结构。我项目中用得最多的是基于链表的多生产者多消费者（MPMC）队列。

单生产者单消费者（SPSC）队列

这种最简单，甚至不需要CAS。只需要两个原子变量：head和tail。生产者只写tail，消费者只读head。我在做音频流处理时就用过这种，延迟极低。

template<typename T, size_t Capacity>
class SPSCQueue {
    std::atomic<size_t> head_{0};
    std::atomic<size_t> tail_{0};
    T buffer_[Capacity];
    
public:
    bool push(const T& item) {
        size_t tail = tail_.load(std::memory_order_relaxed);
        size_t next = (tail + 1) % Capacity;
        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, std::memory_order_release);
        return true;
    }
};

经验之谈：SPSC队列的memory_order选择很关键。release-acquire语义保证了生产者写入的数据对消费者可见，同时不会引入不必要的同步开销。

多生产者多消费者（MPMC）队列

MPMC队列就复杂多了。我常用的方案是基于链表的Hazard Pointer实现，或者用epoch-based reclamation。这里给一个简化版的CAS实现思路：

// 伪代码：无锁链表队列的push操作
void push(Node* node) {
    while (true) {
        Node* tail = tail_.load();
        Node* next = tail->next.load();
        if (tail == tail_.load()) {  // 检查一致性
            if (next == nullptr) {
                if (tail->next.compare_exchange_weak(next, node)) {
                    tail_.compare_exchange_strong(tail, node);
                    return;
                }
            } else {
                tail_.compare_exchange_strong(tail, next);  // 帮助推进tail
            }
        }
    }
}

注意：无锁队列的ABA问题是个大坑。我曾经在生产环境遇到过，排查了整整两天。解决方案是用带标记的指针（tagged pointer）或Hazard Pointer。

无锁栈（Lock-Free Stack）

无锁栈比队列简单，本质就是一个单向链表的头插法。核心操作是CAS更新栈顶指针。

template<typename T>
class LockFreeStack {
    struct Node {
        T data;
        Node* next;
    };
    std::atomic<Node*> head_{nullptr};
    
public:
    void push(const T& value) {
        Node* node = new Node{value, nullptr};
        node->next = head_.load();
        while (!head_.compare_exchange_weak(node->next, node)) {
            // CAS失败则重试，node->next会被更新为最新的head
        }
    }
    
    bool pop(T& value) {
        Node* old_head = head_.load();
        while (old_head && 
               !head_.compare_exchange_weak(old_head, old_head->next)) {
            // 重试直到成功或栈空
        }
        if (!old_head) return false;
        value = old_head->data;
        // 注意：这里需要处理内存回收！
        return true;
    }
};

嗯，这里要注意内存回收问题。直接delete old_head会导致其他线程还在访问它。我一般用Hazard Pointer或引用计数来解决。

读写分离（Read-Write Separation）

读写分离的思路很简单：读操作不加锁，写操作通过拷贝替换来实现。适用于读多写少的场景，比如配置中心、路由表。

我做过一个实时风控系统，规则表每秒更新一次，但被查询几万次。用读写分离后，读性能几乎和单线程一样快。

template<typename T>
class ReadWriteSeparated {
    std::atomic<T*> data_{nullptr};
    std::mutex write_mutex_;
    
public:
    const T* read() const {
        return data_.load(std::memory_order_acquire);
    }
    
    void write(const T& new_data) {
        T* copy = new T(*data_.load());  // 拷贝当前数据
        *copy = new_data;                // 修改拷贝
        T* old = data_.exchange(copy, std::memory_order_release);
        // 注意：旧数据的回收需要延迟处理
    }
};

关键点：写操作必须保证在读操作完成之前，旧数据不会被释放。否则读线程可能访问到野指针。

RCU（Read-Copy-Update）

RCU是Linux内核里广泛使用的同步机制，在用户态C++中也有实现。它的核心思想是：读操作完全无锁，写操作通过发布-订阅机制来更新。

RCU的工作流程：

读端：进入临界区时标记自己正在读，退出时取消标记
写端：拷贝数据，修改拷贝，然后原子替换指针
回收：等待所有在读的线程退出临界区后，再释放旧数据

class RCUProtectedData {
    std::atomic<int*> data_{nullptr};
    // 每个线程的读端状态（简化示意）
    thread_local static bool in_read_ = false;
    
public:
    int read() {
        in_read_ = true;
        int* ptr = data_.load(std::memory_order_acquire);
        int val = *ptr;
        in_read_ = false;
        return val;
    }
    
    void update(int new_val) {
        int* old = data_.load();
        int* new_ptr = new int(*old);
        *new_ptr = new_val;
        data_.store(new_ptr, std::memory_order_release);
        // 等待所有读线程退出
        synchronize_rcu();
        delete old;
    }
};

我的建议：RCU适合读操作远多于写操作的场景。如果写操作频繁，RCU的等待开销会很大。我一般在配置更新、路由表这类场景用RCU。

知识体系总览

下面这张图总结了四种并发数据结构的核心逻辑和适用场景：

避坑指南

我曾经在一个金融交易系统中使用无锁队列，上线后偶尔出现数据丢失。排查后发现是ABA问题导致的节点被错误回收。从那以后，我养成了几个习惯：

永远不要假设CAS一定成功：一定要在循环中重试
内存回收比数据操作更难：Hazard Pointer、Epoch-Based Reclamation、引用计数，选一个用熟
测试要覆盖极端并发：单线程测试通过不代表并发没问题
性能分析要量化：不要凭感觉优化，用perf、火焰图说话

重要提醒：无锁数据结构不是银弹。如果临界区代码较长，或者写操作频繁，传统锁可能更合适。我见过太多人为了炫技而用无锁，结果性能反而更差。

好了，并发数据结构这块就讲到这里。每种方案都有其适用场景，关键是要理解它们的原理和权衡。实际项目中，我建议先从简单的读写分离或SPSC队列开始，等真正遇到性能瓶颈时再考虑更复杂的无锁实现。

📚 课程目录

并发数据结构：无锁队列、无锁栈、读写分离、RCU

为什么需要无锁数据结构？

无锁队列（Lock-Free Queue）

单生产者单消费者（SPSC）队列

多生产者多消费者（MPMC）队列

无锁栈（Lock-Free Stack）

读写分离（Read-Write Separation）

RCU（Read-Copy-Update）

知识体系总览

避坑指南