4. 无锁数据结构（下）：多生产者多消费者（MPMC）队列、无锁栈与无锁哈希表、性能对比与适用场景

好，咱们接着聊无锁数据结构。上一章我们把单生产者单消费者（SPSC）队列讲透了，那玩意儿在特定场景下确实快得飞起。但现实世界哪有那么理想？你想想看，一个高频交易引擎里，多个行情线程往队列里塞数据，多个策略线程同时往外取——这就是典型的多生产者多消费者（MPMC）场景。

我个人习惯，在真正动手写代码之前，先想清楚一个问题：我到底需不需要MPMC？ 很多时候，一个SPSC队列加一个分发线程就能解决问题，何必自找麻烦？但如果你确实需要多个线程同时读写，那咱们今天就来硬碰硬。

4.1 MPMC队列：最常用的无锁数据结构

MPMC队列，说白了就是允许多个线程同时往里push，多个线程同时往外pop。这玩意儿在交易系统里太常见了——比如多个行情源同时写入订单簿更新，多个策略引擎同时消费这些更新。

我在项目中遇到过最头疼的问题：用了一个有锁的MPMC队列，结果在高频行情下，锁竞争导致延迟抖动从微秒级跳到了毫秒级。那次教训让我下定决心，必须上无锁方案。

核心要点：MPMC队列的实现难点在于，多个生产者同时修改头指针，多个消费者同时修改尾指针，这需要原子操作和内存序的精确配合。

下面是一个基于数组的MPMC队列核心实现：

template<typename T, size_t Capacity>
class MPMCQueue {
    static_assert(Capacity >= 2, "Capacity must be at least 2");
    static_assert((Capacity & (Capacity - 1)) == 0, "Capacity must be power of 2");
    
    struct Cell {
        std::atomic<size_t> sequence;
        T data;
    };
    
    Cell buffer[Capacity];
    alignas(64) std::atomic<size_t> enqueue_pos{0};
    alignas(64) std::atomic<size_t> dequeue_pos{0};
    
public:
    MPMCQueue() {
        for (size_t i = 0; i < Capacity; ++i) {
            buffer[i].sequence.store(i, std::memory_order_relaxed);
        }
    }
    
    bool push(const T& data) {
        size_t pos = enqueue_pos.load(std::memory_order_relaxed);
        Cell* cell;
        size_t seq;
        
        while (true) {
            cell = &buffer[pos & (Capacity - 1)];
            seq = cell->sequence.load(std::memory_order_acquire);
            intptr_t diff = static_cast<intptr_t>(seq) - static_cast<intptr_t>(pos);
            
            if (diff == 0) {
                // 槽位可用，尝试CAS获取
                if (enqueue_pos.compare_exchange_weak(pos, pos + 1, 
                    std::memory_order_relaxed)) {
                    break;
                }
            } else if (diff < 0) {
                // 队列满
                return false;
            } else {
                // 其他生产者正在操作，自旋等待
                pos = enqueue_pos.load(std::memory_order_relaxed);
            }
        }
        
        cell->data = data;
        cell->sequence.store(pos + 1, std::memory_order_release);
        return true;
    }
    
    bool pop(T& data) {
        size_t pos = dequeue_pos.load(std::memory_order_relaxed);
        Cell* cell;
        size_t seq;
        
        while (true) {
            cell = &buffer[pos & (Capacity - 1)];
            seq = cell->sequence.load(std::memory_order_acquire);
            intptr_t diff = static_cast<intptr_t>(seq) - static_cast<intptr_t>(pos + 1);
            
            if (diff == 0) {
                if (dequeue_pos.compare_exchange_weak(pos, pos + 1,
                    std::memory_order_relaxed)) {
                    break;
                }
            } else if (diff < 0) {
                return false;
            } else {
                pos = dequeue_pos.load(std::memory_order_relaxed);
            }
        }
        
        data = cell->data;
        cell->sequence.store(pos + Capacity, std::memory_order_release);
        return true;
    }
};

避坑指南：我曾经在内存序上栽过大跟头。一开始图省事全用memory_order_seq_cst，结果性能比有锁还差。后来改成acquire/release配对，性能直接翻倍。记住：能不用seq_cst就别用。

4.2 无锁栈：简单但有限

无锁栈的实现比队列简单得多，但适用场景也窄。说白了，它就是个后进先出（LIFO）的结构。在交易系统里，我主要用它来管理内存池——回收和分配固定大小的对象。

为什么说它有限？因为栈的竞争点只有一个头指针，在高并发下CAS失败的几率很高。你想想看，十个线程同时push，只有一个能成功，其他九个都得重试。这在延迟敏感的场合是致命的。

template<typename T>
class LockFreeStack {
    struct Node {
        T data;
        Node* next;
    };
    
    std::atomic<Node*> head{nullptr};
    
public:
    void push(const T& data) {
        Node* node = new Node{data, nullptr};
        node->next = head.load(std::memory_order_relaxed);
        
        while (!head.compare_exchange_weak(node->next, node,
            std::memory_order_release, std::memory_order_relaxed)) {
            // CAS失败，node->next已被更新为最新head
        }
    }
    
    bool pop(T& data) {
        Node* node = head.load(std::memory_order_relaxed);
        
        while (node && !head.compare_exchange_weak(node, node->next,
            std::memory_order_acquire, std::memory_order_relaxed)) {
            // 重试
        }
        
        if (!node) return false;
        data = node->data;
        delete node;
        return true;
    }
};

注意：上面的实现有个ABA问题。在高并发下，一个节点被释放后又被重新分配，指针值相同但内容变了。解决方案是使用带标记的指针（tagged pointer）或hazard pointer。我个人更推荐用RCU（Read-Copy-Update）来管理内存回收。

4.3 无锁哈希表：高性能查找的利器

无锁哈希表是这三个里面最复杂的。在交易系统里，我拿它来做订单簿的快速查找——比如根据订单ID找到对应的订单信息。哈希表的核心挑战在于：多个线程同时插入、删除、查找，如何保证一致性且不阻塞？

我建议采用开放寻址法加版本号的方案。每个槽位包含一个原子版本号，通过CAS来原子地更新键值对。查找时，如果版本号不匹配，说明数据正在被修改，自旋等待即可。

template<typename K, typename V, size_t Capacity>
class LockFreeHashMap {
    struct Entry {
        std::atomic<uint64_t> version{0};  // 高32位：状态，低32位：版本
        K key;
        V value;
    };
    
    Entry table[Capacity];
    
    static constexpr uint64_t EMPTY = 0;
    static constexpr uint64_t WRITING = 1ULL << 32;
    
public:
    bool insert(const K& key, const V& value) {
        size_t idx = hash(key) & (Capacity - 1);
        
        for (size_t i = 0; i < Capacity; ++i) {
            size_t pos = (idx + i) & (Capacity - 1);
            Entry& entry = table[pos];
            uint64_t ver = entry.version.load(std::memory_order_acquire);
            
            if (ver == EMPTY) {
                uint64_t new_ver = WRITING | 1;
                if (entry.version.compare_exchange_strong(ver, new_ver,
                    std::memory_order_acquire)) {
                    entry.key = key;
                    entry.value = value;
                    entry.version.store(1, std::memory_order_release);
                    return true;
                }
            }
        }
        return false;  // 表满
    }
    
    bool find(const K& key, V& value) {
        size_t idx = hash(key) & (Capacity - 1);
        
        for (size_t i = 0; i < Capacity; ++i) {
            size_t pos = (idx + i) & (Capacity - 1);
            Entry& entry = table[pos];
            uint64_t ver = entry.version.load(std::memory_order_acquire);
            
            if (ver == EMPTY) return false;
            if ((ver & WRITING) == 0 && entry.key == key) {
                value = entry.value;
                return true;
            }
        }
        return false;
    }
};

4.4 性能对比与适用场景

好了，三种数据结构都讲完了。咱们来做个对比，方便你选型时参考。

数据结构 吞吐量 延迟抖动 实现复杂度 适用场景
MPMC队列 高 低 中 多线程数据传递、事件分发
无锁栈 中 高（CAS竞争） 低 内存池、对象回收
无锁哈希表 中高 中（取决于负载因子） 高 快速查找、缓存

我的建议：

  • 如果只是数据传递，优先用MPMC队列。它最成熟，坑最少。
  • 无锁栈只用在低竞争场景，比如每个线程有自己的回收池。
  • 无锁哈希表适合读多写少的场景。写操作多的话，考虑分段锁。

最后，我想说一句：无锁不是银弹。它只是工具，用对了地方是神器，用错了就是灾难。我见过太多人为了炫技而用无锁，结果性能反而更差。记住：先测量，再优化。

无锁数据结构选型决策树 无锁数据结构 数据传递/事件分发 内存池/对象回收 MPMC队列 无锁栈 快速查找/缓存 无锁哈希表 注：箭头方向表示选型决策路径，实际选择需结合具体场景和性能测试

专注资料整理

← 上一章 📖 返回目录 下一章 →