c++怎么实现一个无锁队列_c++无锁队列(lock-free queue)的实现原理

新闻中心 NEWS CENTER

您当前位置：首页 > 新闻中心 > 网络学院

c++怎么实现一个无锁队列_c++无锁队列(lock-free queue)的实现原理

2025-11-02

浏览次数：次

返回列表

无锁队列通过原子操作和CAS实现多线程安全，避免互斥锁开销。核心是使用std::atomic与compare_exchange_weak/strong保证指针更新的原子性，典型结构包括SPSC数组队列和Michael & Scott链表算法。关键挑战为ABA问题与内存回收，需用版本号或Hazard Pointer等机制解决。

c++怎么实现一个无锁队列_c++无锁队列(lock-free queue)的实现原理

实现一个无锁队列（Lock-Free Queue）的关键在于利用原子操作（atomic operations）和内存顺序（memory ordering）来避免使用互斥锁，从而提升多线程环境下的性能。C++ 中可以通过 std::atomic 和 CAS（Compare-And-Swap）操作来实现。

基本原理：CAS 与原子指针操作

无锁队列的核心是使用原子操作来安全地修改共享数据结构，而不需要加锁。最常用的操作是 compare_exchange_weak 或 compare_exchange_strong，它们能比较并交换指针值，保证在多线程竞争时只有一个线程能成功更新。

一个典型的无锁队列基于链表结构，包含头指针（head）和尾指针（tail），所有操作都通过原子方式更新这两个指针。

主要挑战包括：

A-B-A 问题：某个指针被修改后又恢复原值，导致 CAS 误判。可通过加入版本号（如 double-wide CAS，使用 tagged pointer）解决。
内存回收困难：节点被出队后不能立即 delete，因为其他线程可能还持有指针。可使用 Hazard Pointer、RCU 或延迟释放机制处理。
ABA 问题的简化方案：某些实现中通过不重用节点或使用内存池规避。

简易单生产者单消费者无锁队列实现

在特定场景下（如 SPSC），可以简化实现。以下是一个基于循环数组的 SPSC 无锁队列示例：

template<typename T, size_t Size>
class LockFreeQueueSPSC {
    T buffer[Size];
    std::atomic<size_t> head{0}; // 入队位置
    std::atomic<size_t> tail{0}; // 出队位置
<p>public:
bool enqueue(const T& value) {
size_t current_tail = tail.load();
size_t next_tail = (current_tail + 1) % Size;
if (next_tail == head.load()) {
return false; // 队列满
}
buffer[current_tail] = value;
tail.store(next_tail);
return true;
}</p><pre class='brush:php;toolbar:false;'>bool dequeue(T& result) {
    size_t current_head = head.load();
    if (current_head == tail.load()) {
        return false; // 队列空
    }
    result = buffer[current_head];
    size_t next_head = (current_head + 1) % Size;
    head.store(next_head);
    return true;
}

};

这个版本适用于单生产者单消费者场景，无需强内存序，性能高。但 SMP（多生产者多消费者）场景需要更复杂的设计。

Musho

AI网页设计Figma插件

76 查看详情 Musho

多生产者多消费者无锁队列（Michael & Scott 算法）

这是经典的无锁队列算法，使用链表结构，每个节点包含数据和指向下一个节点的指针。

template<typename T>
class LockFreeQueueMPSC {
struct Node {
    T data;
    std::atomic<Node*> next;
<pre class='brush:php;toolbar:false;'>Node() : next(nullptr) {}
Node(const T& d) : data(d), next(nullptr) {}

};

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

public: LockFreeQueueMPSC() { Node* dummy = new Node(); head.store(dummy); tail.store(dummy); }

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

    while (true) {
        old_tail = tail.load();
        old_next = old_tail->next.load();

        if (old_tail == tail.load()) { // 检查是否被其他线程修改
            if (old_next == nullptr) {
                if (old_tail->next.compare_exchange_weak(old_next, new_node)) {
                    tail.compare_exchange_weak(old_tail, new_node);
                    return;
                }
            } else {
                // 队尾未更新，推进 tail
                tail.compare_exchange_weak(old_tail, old_next);
            }
        }
    }
}

bool dequeue(T& result) {
    Node* old_head = nullptr;
    Node* old_tail = nullptr;
    Node* old_next = nullptr;

    while (true) {
        old_head = head.load();
        old_tail = tail.load();
        old_next = old_head->next.load();

        if (old_head == head.load()) {
            if (old_head == old_tail) {
                if (old_next == nullptr) {
                    return false; // 队列为空
                }
                // tail 落后，尝试推进
                tail.compare_exchange_weak(old_tail, old_next);
            } else {
                result = old_next->data;
                if (head.compare_exchange_weak(old_head, old_next)) {
                    delete old_head; // 注意：此处有内存回收风险
                    return true;
                }
            }
        }
    }
}

};

该实现遵循 Michael & Scott 算法逻辑，但在真实环境中需处理内存回收问题。直接 delete 节点可能导致其他线程访问已释放内存。

注意事项与优化建议

无锁队列虽然高效，但实现复杂，需注意以下几点：

内存顺序：默认使用 memory_order_seq_cst 最安全，但可依据场景调整为 memory_order_acquire/release 提升性能。
ABA 问题：在通用场景中，建议使用带版本号的指针（如 struct { Node* ptr; int version; }），配合 128 位 CAS（如有硬件支持）。
内存回收：推荐使用 Hazard Pointer 或 epoch-based reclamation 机制安全释放节点。
性能权衡：无锁不等于高性能，高竞争下可能因反复重试导致 CPU 占用高。