C++原子操作实现无锁队列详解

无锁队列通过原子操作实现线程安全的并发访问，使用 std::atomic 管理 head 和 tail 指针，结合内存序控制与虚拟头节点简化边界处理，在高并发下需解决 ABA 问题与内存回收难题。

实现一个无锁队列（lock-free queue）需要利用 C++ 的原子操作（atomic operations）来避免使用互斥锁，从而提升多线程环境下的性能。核心思想是通过原子指针操作和内存顺序控制，确保多个线程可以安全地并发执行入队（enqueue）和出队（dequeue）操作。

基本原理：基于链表的无锁队列

最常见的无锁队列实现是基于单向链表的结构，使用两个原子指针：head 和 tail，分别指向队列的头和尾。每个节点包含数据和指向下一个节点的指针。

关键点在于所有对指针的操作都必须是原子的，并且要正确处理 ABA 问题和内存重排序。

以下是一个简化但可运行的无锁队列实现：

#include <atomic>
#include <memory>

template<typename T>
class LockFreeQueue {
private:
    struct Node {
        std::shared_ptr<T> data;
        std::atomic<Node*> next;

        Node() : data(nullptr), next(nullptr) {}

        Node(const T& d) : data(std::make_shared<T>(d)), next(nullptr) {}
    };

    std::atomic<Node*> head;
    std::atomic<Node*> tail;

    // 辅助函数：尝试释放已出队的节点
    void free_if_need(Node* old_head) {
        if (old_head) {
            delete old_head;
        }
    }

public:
    LockFreeQueue() {
        Node* dummy = new Node();
        head.store(dummy, std::memory_order_relaxed);
        tail.store(dummy, std::memory_order_relaxed);
    }

    ~LockFreeQueue() {
        while (Node* h = head.load()) {
            head.store(h->next.load());
            delete h;
        }
    }

    void enqueue(const T& data) {
        Node* new_node = new Node(data);
        Node* old_tail = nullptr;
        Node* next = nullptr;

        while (true) {
            old_tail = tail.load(std::memory_order_acquire);
            next = old_tail->next.load(std::memory_order_acquire);

            // 检查 tail 是否滞后
            if (old_tail != tail.load(std::memory_order_acquire)) {
                continue; // 重新尝试
            }

            // 如果 tail 没有指向最后一个节点，尝试推进 tail
            if (next != nullptr) {
                tail.compare_exchange_weak(old_tail, next, std::memory_order_release, std::memory_order_acquire);
                continue;
            }

            // 尝试将新节点链接到 tail 后
            if (old_tail->next.compare_exchange_weak(next, new_node, std::memory_order_release, std::memory_order_acquire)) {
                break; // 成功链接
            }
        }

        // 尝试更新 tail 指向新节点
        tail.compare_exchange_weak(old_tail, new_node, std::memory_order_release, std::memory_order_acquire);
    }

    std::shared_ptr<T> dequeue() {
        Node* old_head = nullptr;
        while (true) {
            old_head = head.load(std::memory_order_acquire);
            Node* old_tail = tail.load(std::memory_order_acquire);
            Node* next = old_head->next.load(std::memory_order_acquire);

            // 判断队列是否为空
            if (old_head == old_tail) {
                if (next == nullptr) {
                    return std::shared_ptr<T>(); // 队列空
                }
                // tail 滞后，尝试推进
                tail.compare_exchange_weak(old_tail, next, std::memory_order_release, std::memory_order_acquire);
                continue;
            }

            // 读取数据并尝试移动 head
            if (head.compare_exchange_weak(old_head, next, std::memory_order_release, std::memory_order_acquire)) {
                std::shared_ptr<T> res = next->data;
                // 延迟释放 old_head（注意：生产环境应使用 hazard pointer 或 RCU）
                free_if_need(old_head);
                return res;
            }
        }
    }
};

关键点解析

1. 原子指针操作： 使用 `std::atomic` 来保证 head 和 tail 的读写是原子的，防止数据竞争。

2. 内存顺序（memory order）：

• memory_order_acquire 用于读操作，确保后续读写不会被重排到该操作之前。
• memory_order_release 用于写操作，确保前面的读写不会被重排到该操作之后。
• compare_exchange_weak 在循环中使用，允许失败后重试。

3. 虚拟头节点（dummy node）： 构造时创建一个空节点作为初始 head 和 tail，简化边界判断。

4. tail 滞后处理： 其他线程可能在修改 next，但还没更新 tail，此时需要帮助完成 tail 推进。

存在的问题与改进方向

上述实现虽然能工作，但在高并发下仍可能遇到问题：

• ABA 问题： 虽然指针值没变，但中间可能已被释放并重用。可通过双字 CAS（如 __int128 包装指针+计数器）解决。
• 内存回收困难： 直接 delete 可能导致其他线程访问已释放内存。应使用 Hazard Pointer、RCU 或 epoch-based reclamation。
• 性能： 多线程竞争 tail 时可能频繁失败。可考虑 Michael-Scott 队列的优化版本。

使用示例

std::vector<std::thread> producers;
std::vector<std::thread> consumers;

for (int i = 0; i < 2; ++i) {
    producers.emplace_back([&queue, i] {
        for (int j = 0; j < 1000; ++j) {
            queue.enqueue(i * 1000 + j);
        }
    });
}

for (int i = 0; i < 2; ++i) {
    consumers.emplace_back([&queue] {
        for (int j = 0; j < 1000; ++j) {
            auto val = queue.dequeue();
            if (val) {
                std::cout << "Dequeued: " << *val << std::endl;
            }
        }
    });
}

for (auto& t : producers) t.join();
for (auto& t : consumers) t.join();

return 0;

基本上就这些。无锁队列实现复杂，调试困难，建议在真正需要极致性能时才使用，否则优先考虑互斥锁 + 条件变量的方案。