C++如何实现一个无锁队列_C++原子操作与CAS原理实现高性能并发数据结构-C++-PHP中文网

C++如何实现一个无锁队列_C++原子操作与CAS原理实现高性能并发数据结构

尼克

发布： 2025-12-21 19:35:02

原创

420人浏览过

无锁队列通过原子操作和CAS实现多线程并发访问，使用std::atomic和内存序优化性能，需解决ABA问题并谨慎处理内存回收。

c++如何实现一个无锁队列_c++原子操作与cas原理实现高性能并发数据结构

实现一个无锁队列（lock-free queue）是高性能并发编程中的常见需求。相比使用互斥量（mutex）保护的队列，无锁队列通过原子操作和CAS（Compare-And-Swap）机制避免线程阻塞，提升多线程环境下的吞吐量。C++ 提供了 std::atomic 和底层原子指令支持，使得实现无锁数据结构成为可能。

无锁队列的基本原理

无锁队列的核心思想是：多个线程可以同时访问共享数据结构，但通过原子操作保证操作的完整性，而不是依赖锁来串行化访问。关键在于使用 CAS（Compare-And-Swap） 操作来更新指针或状态变量。

CAS 的逻辑是：如果当前值等于预期值，则将其更新为新值，否则不做修改。这个过程是原子的，由硬件层面保障。在 C++ 中，可以通过 std::atomic.compare_exchange_weak() 或 compare_exchange_strong() 实现。

基于链表的无锁队列实现

一个常见的无锁队列实现是使用单向链表，维护 head 和 tail 两个指针。每个节点包含数据和指向下一个节点的指针。入队操作从 tail 插入，出队操作从 head 移除。

立即学习“C++免费学习笔记（深入）”；

以下是简化版的无锁队列实现：

#include <atomic>
#include <memory>
<p>template<typename T>
class LockFreeQueue {
private:
struct Node {
T data;
std::atomic<Node*> next;
Node(const T& d) : data(d), next(nullptr) {}
};</p><pre class='brush:php;toolbar:false;'>std::atomic<Node*> head;
std::atomic<Node*> tail;

登录后复制

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

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

    while (true) {
        expected_tail = tail.load(std::memory_order_acquire);
        // 尝试将 tail 的 next 设置为新节点
        if (expected_tail->next.compare_exchange_weak(
                null_next, new_node, std::memory_order_release)) {
            // 更新 tail 指针
            tail.compare_exchange_weak(expected_tail, new_node,
                                       std::memory_order_acq_rel);
            break;
        } else {
            // tail 已被其他线程更新，尝试更新本地副本
            tail.compare_exchange_weak(expected_tail, expected_tail->next.load(),
                                       std::memory_order_acq_rel);
        }
    }
}

bool dequeue(T& result) {
    Node* old_head = head.load(std::memory_order_acquire);
    Node* old_tail = tail.load(std::memory_order_acquire);
    Node* next_head = old_head->next.load(std::memory_order_acquire);

    if (old_head == old_tail) {
        if (next_head == nullptr) {
            return false; // 队列为空
        }
        // tail 落后，尝试推进
        tail.compare_exchange_weak(old_tail, next_head, std::memory_order_acq_rel);
        return false;
    }

    if (head.compare_exchange_weak(old_head, next_head, std::memory_order_acq_rel)) {
        result = next_head->data;
        delete old_head; // 注意：存在 ABA 问题风险
        return true;
    }
    return false; // 失败，重试
}

~LockFreeQueue() {
    while (dequeue(T{}));
    delete head.load();
}

登录后复制

};