无锁队列通过原子操作和CAS实现多线程安全，避免互斥锁开销。核心是使用std::atomic与compare_exchange_weak/strong保证指针更新的原子性，典型结构包括SPSC数组队列和Michael & Scott链表算法。关键挑战为ABA问题与内存回收，需用版本号或Hazard Pointer等机制解决。
实现一个无锁队列（Lock-Free Queue）的关键在于利用原子操作（atomic operations）和内存顺序（memory ordering）来避免使用互斥锁，从而提升多线程环境下的性能。c++ 中可以通过 std::atomic 和 CAS（Compare-And-Swap）操作来实现。
基本原理：CAS 与原子指针操作
无锁队列的核心是使用原子操作来安全地修改共享数据结构，而不需要加锁。最常用的操作是 compare_exchange_weak 或 compare_exchange_strong，它们能比较并交换指针值，保证在多线程竞争时只有一个线程能成功更新。
一个典型的无锁队列基于链表结构，包含头指针（head）和尾指针（tail），所有操作都通过原子方式更新这两个指针。
主要挑战包括：
立即学习“C++免费学习笔记（深入）”；
- 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（多生产者多消费者）场景需要更复杂的设计。
多生产者多消费者无锁队列（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 占用高。
基本上就这些。无锁队列适合对延迟敏感、锁竞争激烈的场景，但开发难度大，建议优先考虑成熟的库如 absl::Mutex、folly::MPMCQueue 或 boost::lockfree::queue。