C++ 无锁队列（Lock-Free Queue）详解

解决方案

• 标记指针（Tagged Pointer）：给指针附加一个版本号（如 std::atomic<std::pair<Node*, uint64_t>>），每次修改指针时版本号递增。CAS 时同时检查指针和版本号，即使指针相同，版本号不同也会拒绝更新。
• 风险指针（Hazard Pointers）：线程访问节点前，将节点指针存入'风险指针'，确保节点被访问期间不会被回收。
• 基于时代的回收（Epoch-Based Reclamation）：按'时代'划分线程操作，仅回收所有线程都已退出该时代的节点。

2.2 内存回收问题

无锁队列的节点是动态分配的，但不能直接 delete——可能有其他线程仍在访问该节点（如 T1 读取节点后，T2 删除节点并 delete，T1 后续访问会触发野指针）。

解决方案与 ABA 问题重叠（风险指针、时代回收），核心思想是延迟回收，确保节点不再被任何线程引用后再释放。

2.3 虚假共享（False Sharing）

队列的 head 和 tail 原子变量若位于同一缓存行（64 字节），多线程并发修改时会导致缓存行频繁失效（缓存颠簸），严重影响性能。

解决方案：

• 缓存行对齐：用 std::hardware_destructive_interference_size（C++17）将 head 和 tail 分到不同缓存行。
• 填充无用数据：在 head 和 tail 之间插入足够的占位字节（如 64 字节），强制编译器分开存储。

2.4 多生产者/多消费者（MPMC）同步

• 单生产者单消费者（SPSC）：无锁队列实现最简单，无需处理 head/tail 的并发修改冲突。
• 多生产者多消费者（MPMC）：需通过 CAS 保证 head（出队）和 tail（入队）的原子修改，逻辑更复杂。

三、常见无锁队列实现结构

3.1 Michael-Scott 无锁队列（MPMC，链表型）

由 Michael 和 Scott 于 1996 年提出，是最经典的 MPMC 无锁队列实现，基于单向链表，核心是通过 CAS 原子修改 head 和 tail。

结构设计

template<typename T>
struct Node {
    T data;
    std::atomic<Node*> next; // 原子指针，指向后继节点
    Node(const T& val) : data(val), next(nullptr) {}
};

template<typename T>
class MSQueue {
private:
    // 标记指针：Node* + 版本号（解决 ABA 问题）
    struct TaggedPtr {
        Node<T>* ptr;
        uint64_t tag;
        TaggedPtr(Node<T>* p = nullptr, uint64_t t = 0) : ptr(p), tag(t) {}
        bool operator==(const TaggedPtr& other) const {
            return ptr == other.ptr && tag == other.tag;
        }
    };
    std::atomic<TaggedPtr> head; // 队列头（原子标记指针）
    std::atomic<TaggedPtr> tail; // 队列尾（原子标记指针）
    // 缓存行对齐，避免虚假共享
    char padding[std::hardware_destructive_interference_size - sizeof(std::atomic<TaggedPtr>)];
public:
    MSQueue() {
        Node<T>* dummy = new Node<T>(T{}); // 哨兵节点（简化入队/出队逻辑）
        head.store({dummy, 0}, std::memory_order_relaxed);
        tail.store({dummy, 0}, std::memory_order_relaxed);
    }
    ~MSQueue(); // 需遍历释放所有节点
};

核心操作：入队（enqueue）

1. 创建新节点；
2. 循环通过 CAS 更新 tail：
   • 读取当前 tail（acquire 内存序，确保看到最新的 tail->next）；
   • 若 tail->ptr->next 为 nullptr，尝试将其 CAS 改为新节点（release 内存序）；
   • 若 CAS 成功，再将 tail CAS 改为新节点（release 内存序），入队完成；
   • 若失败（其他线程已修改 tail 或 tail->next），重试。

void enqueue(const T& val) {
    Node<T>* new_node = new Node<T>(val);
    TaggedPtr new_tail{new_node, 0};
    while (true) {
        TaggedPtr current_tail = tail.load(std::memory_order_acquire); // 读 tail，acquire
        Node<T>* tail_ptr = current_tail.ptr;
        std::atomic<Node*>& next = tail_ptr->next;
        // 检查是否有其他线程已更新 tail 的 next
        Node<T>* null_ptr = nullptr;
        if (next.compare_exchange_strong(
                null_ptr, new_node,
                std::memory_order_release, // 写 next，release
                std::memory_order_relaxed)) {
            // 成功将新节点接在 tail 后，更新 tail
            TaggedPtr expected_tail{tail_ptr, current_tail.tag};
            new_tail.tag = current_tail.tag + 1; // 版本号递增
            tail.compare_exchange_weak(
                expected_tail, new_tail,
                std::memory_order_release, // 写 tail，release
                std::memory_order_relaxed);
            return;
        } else {
            // 其他线程已修改 next，尝试帮助更新 tail（优化：减少重试次数）
            Node<T>* next_ptr = next.load(std::memory_order_relaxed);
            TaggedPtr expected_tail{tail_ptr, current_tail.tag};
            TaggedPtr candidate_tail{next_ptr, current_tail.tag + 1};
            tail.compare_exchange_weak(
                expected_tail, candidate_tail,
                std::memory_order_release,
                std::memory_order_relaxed);
        }
    }
}

核心操作：出队（dequeue）

1. 循环通过 CAS 更新 head：
   • 读取当前 head 和 tail（acquire 内存序）；
   • 若队列空（head->ptr == tail->ptr 且 head->ptr->next 为 nullptr），返回失败；
   • 读取 head->ptr->next（实际数据节点）；
   • 尝试将 head CAS 改为该数据节点（版本号递增，release 内存序）；
   • 若 CAS 成功，释放原 head->ptr（哨兵节点），返回数据；
   • 若失败，重试。

bool dequeue(T& val) {
    while (true) {
        TaggedPtr current_head = head.load(std::memory_order_acquire); // 读 head，acquire
        TaggedPtr current_tail = tail.load(std::memory_order_acquire); // 读 tail，acquire
        Node<T>* head_ptr = current_head.ptr;
        Node<T>* tail_ptr = current_tail.ptr;
        Node<T>* next_ptr = head_ptr->next.load(std::memory_order_relaxed); // 检查队列是否为空，或 head/tail 是否已被其他线程修改
        if (current_head == head.load(std::memory_order_relaxed)) {
            if (head_ptr == tail_ptr) {
                if (next_ptr == nullptr) {
                    return false; // 队列空
                }
                // 帮助更新 tail（入队未完成 tail 更新）
                TaggedPtr expected_tail{tail_ptr, current_tail.tag};
                TaggedPtr candidate_tail{next_ptr, current_tail.tag + 1};
                tail.compare_exchange_weak(
                    expected_tail, candidate_tail,
                    std::memory_order_release,
                    std::memory_order_relaxed);
            } else {
                if (next_ptr == nullptr) {
                    continue; // 中间状态，重试
                }
                // 读取数据（需保证 next_ptr 的数据已完全写入）
                val = next_ptr->data;
                // CAS 更新 head 为 next_ptr，版本号递增
                TaggedPtr expected_head{head_ptr, current_head.tag};
                TaggedPtr new_head{next_ptr, current_head.tag + 1};
                if (head.compare_exchange_strong(
                        expected_head, new_head,
                        std::memory_order_release, // 写 head，release
                        std::memory_order_relaxed)) {
                    delete head_ptr; // 释放原哨兵节点（简化版，未处理内存回收问题）
                    return true;
                }
            }
        }
    }
}

注意

上述是简化版实现，未完全解决内存回收（如 delete head_ptr 可能导致其他线程访问失效节点）和复杂 ABA 场景，仅用于理解核心逻辑。生产环境需结合风险指针或时代回收完善。

3.2 环形无锁队列（SPSC/MPMC，数组型）

基于固定大小数组，通过原子索引 head（出队索引）和 tail（入队索引）实现，适合数据量固定、高吞吐场景。

核心特点

• 单生产者单消费者（SPSC）：head 仅由消费者修改，tail 仅由生产者修改，无需 CAS，仅需内存序保证可见性，实现最简单、性能最优。
• 多生产者多消费者（MPMC）：head 和 tail 需通过 CAS 原子修改，需处理索引环绕（取模）和满/空判断。

SPSC 环形无锁队列示例

#include <atomic>
#include <vector>

template<typename T, size_t Size>
class SPSCQueue {
private:
    static constexpr size_t CAPACITY = Size;
    std::vector<T> buffer; // 原子索引，用 acquire/release 保证可见性
    std::atomic<size_t> head; // 消费者读取索引（仅消费者修改）
    char padding1[std::hardware_destructive_interference_size - sizeof(std::atomic<size_t>)];
    std::atomic<size_t> tail; // 生产者写入索引（仅生产者修改）
    char padding2[std::hardware_destructive_interference_size - sizeof(std::atomic<size_t>)];
public:
    SPSCQueue() : buffer(CAPACITY), head(0), tail(0) {}
    // 生产者入队（仅单线程调用）
    bool enqueue(const T& val) {
        const size_t current_tail = tail.load(std::memory_order_relaxed);
        const size_t next_tail = (current_tail + 1) % CAPACITY;
        if (next_tail == head.load(std::memory_order_acquire)) {
            return false; // 队列满
        }
        buffer[current_tail] = val;
        tail.store(next_tail, std::memory_order_release); // 写 tail，release
        return true;
    }
    // 消费者出队（仅单线程调用）
    bool dequeue(T& val) {
        const size_t current_head = head.load(std::memory_order_relaxed);
        if (current_head == tail.load(std::memory_order_acquire)) {
            return false; // 队列空
        }
        val = buffer[current_head];
        head.store((current_head + 1) % CAPACITY, std::memory_order_release); // 写 head，release
        return true;
    }
    bool empty() const {
        return head.load(std::memory_order_acquire) == tail.load(std::memory_order_acquire);
    }
    bool full() const {
        const size_t next_tail = (tail.load(std::memory_order_acquire) + 1) % CAPACITY;
        return next_tail == head.load(std::memory_order_acquire);
    }
};

关键说明

• SPSC 队列无需 CAS，因为 head 和 tail 分别由单个线程修改，仅需 acquire/release 保证生产者写入的数据对消费者可见。
• 队列满/空判断：通过 next_tail == head（满）和 head == tail（空），牺牲一个元素空间（避免满和空的判断冲突）。
• 性能极高：数组存储无链表节点的动态分配开销，缓存局部性好，适合高频读写场景（如线程池任务队列、日志采集）。

四、无锁队列的优缺点与适用场景

4.1 优点

1. 低延迟：无锁竞争，避免线程阻塞和上下文切换。
2. 高吞吐：高并发下，原子操作的开销远小于锁的竞争开销。
3. 无死锁风险：不依赖锁，自然避免死锁。

4.2 缺点

1. 实现复杂：需处理 ABA 问题、内存回收、内存序等细节，调试难度大。
2. 低并发场景性能不佳：原子操作的硬件开销（如 CAS 需 CPU 总线锁定）可能高于互斥锁，低并发时无优势。
3. 内存开销：链表型无锁队列的节点动态分配、标记指针的版本号都会增加内存开销。
4. 不保证公平性：线程可能因 CAS 失败频繁重试，导致饥饿（无等待队列可避免）。

4.3 适用场景

• 高并发、低延迟需求：如金融交易、实时数据处理、高性能服务器。
• 多线程高频读写：如线程池任务分发、日志队列、消息队列。
• 避免锁竞争瓶颈：如多生产者多消费者场景，锁队列的竞争会导致吞吐量下降。

4.4 不适用场景

• 低并发、低频率读写：如偶尔的线程间数据传递，有锁队列（std::queue + std::mutex）更简单、性能相当。
• 数据量小且固定：普通数组或链表即可满足，无需复杂的无锁设计。
• 对公平性要求高：无锁队列可能导致部分线程长期重试。

五、C++ 生产级无锁队列库推荐

手动实现无锁队列难度大、易出错，生产环境优先使用成熟库：

1. Boost.Lockfree：
   • 支持 lockfree::queue（MPMC）、lockfree::stack（MPSC）、lockfree::spsc_queue（SPSC）。
   • 跨平台，稳定性高，支持内存回收（基于风险指针）。
2. folly::ConcurrentQueue（Facebook Folly）：
   • MPMC 无锁队列，性能优异，支持批量入队/出队、动态扩容。
   • 依赖 Folly 库（需编译安装），功能丰富（如优先级队列、阻塞/非阻塞模式）。
3. moodycamel::ConcurrentQueue：
   • 轻量级 MPMC 无锁队列，无依赖，单头文件，性能接近 Boost.Lockfree。
   • 支持批量操作、超时等待，适合快速集成。
4. absl::synchronization::BoundedQueue（Google Abseil）：
   • 有界 MPMC 队列，支持无锁模式，集成 Abseil 生态，稳定性高。

示例：

#include <boost/lockfree/queue.hpp>

boost::lockfree::queue<int> q(1024); // 容量 1024 的 MPMC 无锁队列
q.push(42); // 入队
int val;
q.pop(val); // 出队

六、总结

无锁队列是 C++ 高并发编程的核心数据结构，其本质是用原子操作替代锁，用内存序保证数据可见性和指令顺序。关键难点是 ABA 问题和内存回收，解决方案包括标记指针、风险指针、时代回收等。

实际开发中，优先选择成熟的第三方库（如 Boost.Lockfree、moodycamel），避免重复造轮子；若需自定义实现，建议从 SPSC 环形队列入手（逻辑简单、不易出错），再逐步挑战 MPMC 链表队列。

核心要点：

• 无锁 ≠ 无开销，原子操作仍有硬件开销，需根据并发强度选择。
• 内存序是关键，错误的内存序会导致数据竞争和可见性问题。
• 测试至关重要，需通过压力测试、线程 sanitizer（TSAN）检测数据竞争。