如何用C++11的原子CAS实现JD Valois的无锁队列？

咱来帮你把这个经典的无锁队列实现补全并把关键细节讲透——Valois的算法是无锁并发里的标杆，用C++11的原子CAS来落地完全可行，不过你的初始代码里有几个容易踩坑的地方得注意。

首先先回顾下Valois论文里的Enqueue核心逻辑，用伪代码表示就是：

Enqueue(x) {
q = new Record();
q->value = x;
q->next = NULL;
do {
p = tail;
} while( ! CAS(p->next, NULL, q) ); // 1：把新节点挂到队列尾部
CAS(tail, p, q); // 2：尝试更新tail到新节点（失败也不影响正确性）
}

这个逻辑的核心是两步CAS：第一步确保新节点被安全挂到队列的实际尾部，第二步尝试更新tail指针（即使失败，其他线程后续会自动修正tail的位置）。

你初始代码的问题点 #

你的代码片段里有个致命问题：Node * next;不是原子类型！多线程环境下读写这个指针会导致数据竞争，完全不符合无锁算法的安全要求，必须改成std::atomic<Node*> next;才行。另外，直接用原始指针管理内存容易泄漏，建议结合智能指针或者更安全的内存回收机制。

完整的C++11实现代码 #

下面是完善后的模板化无锁队列，包含enqueue、dequeue、安全初始化和清理：

#include <atomic>
#include <memory>

template <typename T>
class LockFreeQueue {
private:
    struct Node {
        std::unique_ptr<T> val; // 用unique_ptr自动管理值的内存
        std::atomic<Node*> next; // 必须是原子指针，避免数据竞争

        // 带值的节点构造函数
        Node(std::unique_ptr<T> value) : val(std::move(value)), next(nullptr) {}
        // 哨兵节点构造函数（空节点，用来简化边界逻辑）
        Node() : next(nullptr) {}
    };

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

public:
    LockFreeQueue() {
        // 初始化时创建一个哨兵节点，让head和tail始终指向有效节点
        Node* sentinel = new Node();
        head.store(sentinel, std::memory_order_release);
        tail.store(sentinel, std::memory_order_release);
    }

    ~LockFreeQueue() {
        // 清理所有节点，确保内存不泄漏
        while (Node* node = head.load(std::memory_order_acquire)) {
            head.store(node->next.load(std::memory_order_acquire));
            delete node;
        }
    }

    // 禁用拷贝和移动，避免并发下的非法操作
    LockFreeQueue(const LockFreeQueue&) = delete;
    LockFreeQueue& operator=(const LockFreeQueue&) = delete;
    LockFreeQueue(LockFreeQueue&&) = delete;
    LockFreeQueue& operator=(LockFreeQueue&&) = delete;

    void enqueue(std::unique_ptr<T> t) {
        Node* new_node = new Node(std::move(t));
        Node* old_tail = nullptr;

        do {
            old_tail = tail.load(std::memory_order_acquire);
            Node* next_node = old_tail->next.load(std::memory_order_acquire);

            // 如果tail已经被其他线程更新，重新获取最新的tail
            if (old_tail != tail.load(std::memory_order_acquire)) {
                continue;
            }

            // 发现tail滞后（next不为空），先帮其他线程更新tail到正确位置
            if (next_node != nullptr) {
                tail.compare_exchange_weak(old_tail, next_node, 
                                           std::memory_order_release, 
                                           std::memory_order_acquire);
                continue;
            }

            // 尝试把新节点挂到当前tail的next，失败则重试
        } while (!old_tail->next.compare_exchange_weak(nullptr, new_node, 
                                                      std::memory_order_release, 
                                                      std::memory_order_acquire));

        // 尝试更新tail到新节点，失败也没关系——其他线程会处理
        tail.compare_exchange_weak(old_tail, new_node, 
                                   std::memory_order_release, 
                                   std::memory_order_acquire);
    }

    std::unique_ptr<T> dequeue() {
        Node* old_head = nullptr;
        std::unique_ptr<T> result;

        do {
            old_head = head.load(std::memory_order_acquire);
            Node* next_node = old_head->next.load(std::memory_order_acquire);

            // 如果head已经被其他线程更新，重新获取
            if (old_head != head.load(std::memory_order_acquire)) {
                continue;
            }

            // 队列为空，返回空指针
            if (next_node == nullptr) {
                return nullptr;
            }

            // 取出节点的值
            result = std::move(next_node->val);

            // 尝试更新head到下一个节点（原head是哨兵节点，更新后新的head是原next_node）
        } while (!head.compare_exchange_weak(old_head, next_node, 
                                            std::memory_order_release, 
                                            std::memory_order_acquire));

        // 释放旧的哨兵节点
        delete old_head;
        return result;
    }
};

核心细节解释 #

1. 哨兵节点：初始化时创建一个空节点，让head和tail永远指向有效节点，彻底避免了处理head/tail为NULL的复杂边界情况，简化了enqueue和dequeue的逻辑。
2. 内存顺序：使用std::memory_order_acquire和std::memory_order_release保证原子操作的可见性，防止编译器和CPU的内存重排序导致的并发问题——acquire确保后续操作能看到原子操作的结果，release确保之前的操作对其他线程可见。
3. ABA问题处理：Valois的基础算法存在ABA风险（比如节点被删除后重新分配内存，CAS时指针值相同但实际不是同一个节点）。如果要解决这个问题，可以使用带版本号的原子指针（比如std::atomic<std::pair<Node*, size_t>>），或者用hazard pointer（危险指针）来管理内存回收，避免节点被过早释放。
4. 内存安全：用std::unique_ptr管理节点的值，自动释放内存；节点本身在dequeue时释放旧的哨兵节点，但在高并发场景下，直接删除节点可能有风险（其他线程可能还在引用），更安全的做法是使用hazard pointer或者epoch-based内存回收机制。
5. enqueue的第二步CAS：这一步即使失败也不影响队列正确性——其他线程在enqueue时发现tail的next不为nullptr，会自动帮我们把tail更新到正确的位置。

内容的提问来源于stack exchange，提问作者calvin