无锁队列（Lock-Free Queue）实现详解
无锁队列是并发编程中的高频面试题，尤其在云基础设施、高性能中间件场景中必考。以下从原理到实现逐层拆解。
一、为什么需要无锁队列
传统互斥锁队列的问题：
- 阻塞：线程拿不到锁就睡眠，上下文切换开销大（~1-5μs）
- 优先级反转：低优先级线程持锁，高优先级线程阻塞
- 死锁风险：多锁场景容易出错
- 不可重入：信号处理函数中不能用锁
无锁队列的核心保证：某个线程的延迟或崩溃不会阻塞其他线程的推进（obstruction-free /lock-free/wait-free 三级）。
二、核心原语
1. CAS（Compare-And-Swap）
plaintext
bool CAS(addr, expected, desired):
if *addr == expected:
*addr = desired
return true
return false
C++11 对应
std::atomic::compare_exchange_weak/strong。2. 内存序（Memory Order）
这是面试深水区，必须讲清楚：
表格
| 内存序 | 含义 | 队列中用途 |
|---|---|---|
relaxed |
只保证原子性，不保证顺序 | 计数器、统计字段 |
acquire |
读之后的操作不能重排到读之前 | 出队读 head |
release |
写之前的操作不能重排到写之后 | 入队写 tail |
acq_rel |
兼具 acquire + release | CAS 操作 |
seq_cst |
全局全序 | 默认，最保守，性能最差 |
关键配对：一个线程
release 写，另一个线程 acquire 读，形成 synchronizes-with 关系，保证可见性。3. ABA 问题
线程 1 读
A，准备 CAS；期间线程 2 改成 B 又改回 A；线程 1 的 CAS 成功但状态已变。解决方案：
- 标记指针（Tagged Pointer）：指针 + 版本号打包成一个 64 位值（利用 64 位地址低几位未使用）
- DCAS（Double CAS）：同时比较指针和版本号
- ** Hazard Pointer / Epoch-based Reclamation**：安全内存回收
三、经典实现：Michael-Scott Queue（1996）
这是最经典的无锁队列算法，Java
ConcurrentLinkedQueue、Boost lockfree::queue 都基于它。面试讲这个最稳妥。算法核心
- 链表结构，
head和tail都是原子指针 - 入队只操作
tail，出队只操作head，两者竞争最小 - 使用 dummy node（哨兵节点） 避免 head/tail 在空队列时重合的边界问题
入队（Enqueue）
plaintext
1. 新建节点 node，next = nullptr
2. 循环：
a. 读取 tail = this->tail
b. 读取 next = tail->next
c. 如果 tail 没变（双重检查）：
- 如果 next == nullptr：
CAS(tail->next, nullptr, node) // 尝试把新节点挂到 tail 后面
成功则 CAS(tail, tail, node) // 推进 tail（可能失败，没关系）
返回
- 否则：
CAS(tail, tail, next) // 帮其他线程推进 tail（辅助机制）
注意：tail 可能滞后。如果入队线程 CAS tail 失败，说明另一个线程已经推进了，下一次循环会读到新的 tail。
出队（Dequeue）
plaintext
1. 循环：
a. 读取 head = this->head
b. 读取 tail = this->tail
c. 读取 next = head->next
d. 如果 head 没变：
- 如果 head == tail：
如果 next == nullptr：队列为空，返回 false
否则：CAS(tail, tail, next) // 帮推进 tail
- 否则：
读取 next->value
如果 CAS(head, head, next) 成功：
释放旧 head（需要安全回收机制）
返回 value
C++ 完整实现（带版本号解决 ABA）
cpp
运行
#include <atomic>
#include <cstdint>
#include <utility>
template <typename T>
class MichaelScottQueue {
private:
struct Node {
T value;
std::atomic<Node*> next;
Node(T v) : value(std::move(v)), next(nullptr) {}
Node() : value{}, next(nullptr) {} // dummy node
};
// 用 64 位打包指针 + 版本号（假设指针高16位可用，实际需根据平台调整）
// 更稳妥的做法是用 std::atomic<std::pair<Node*, uint64_t>> 但不保证原子
// 工业实现通常用 128 位 CAS（cmpxchg16b）或 tagged pointer
struct TaggedPtr {
Node* ptr;
uint64_t tag;
};
std::atomic<Node*> head_;
std::atomic<Node*> tail_;
public:
MichaelScottQueue() {
Node* dummy = new Node();
head_.store(dummy, std::memory_order_relaxed);
tail_.store(dummy, std::memory_order_relaxed);
}
~MichaelScottQueue() {
// 清理所有节点
Node* cur = head_.load(std::memory_order_relaxed);
while (cur) {
Node* next = cur->next.load(std::memory_order_relaxed);
delete cur;
cur = next;
}
}
void enqueue(T value) {
Node* node = new Node(std::move(value));
Node* tail;
while (true) {
tail = tail_.load(std::memory_order_acquire);
Node* next = tail->next.load(std::memory_order_acquire);
// 确认 tail 没被其他线程修改
if (tail != tail_.load(std::memory_order_acquire))
continue;
if (next == nullptr) {
// 尝试把新节点链接到 tail 后面
if (tail->next.compare_exchange_weak(
next, node,
std::memory_order_release,
std::memory_order_relaxed)) {
// 尝试推进 tail，失败也没关系（其他线程或下次会推进）
tail_.compare_exchange_strong(
tail, node,
std::memory_order_release,
std::memory_order_relaxed);
return;
}
} else {
// tail 滞后了，帮其他线程推进
tail_.compare_exchange_strong(
tail, next,
std::memory_order_release,
std::memory_order_relaxed);
}
}
}
bool dequeue(T& out) {
Node* head;
while (true) {
head = head_.load(std::memory_order_acquire);
Node* tail = tail_.load(std::memory_order_acquire);
Node* next = head->next.load(std::memory_order_acquire);
if (head != head_.load(std::memory_order_acquire))
continue;
if (head == tail) {
if (next == nullptr)
return false; // 空队列
// tail 滞后，帮推进
tail_.compare_exchange_strong(
tail, next,
std::memory_order_release,
std::memory_order_relaxed);
} else {
out = std::move(next->value);
// 推进 head，旧 head 变成新的 dummy
if (head_.compare_exchange_weak(
head, next,
std::memory_order_release,
std::memory_order_relaxed)) {
// ⚠️ 这里不能直接 delete head！
// 其他线程可能还在读这个节点（通过旧的 head 指针）
// 需要 Hazard Pointer / Epoch / RCU 等安全回收机制
// delete head; // 危险！
return true;
}
}
}
}
};
四、内存安全回收（面试必问）
上面代码中
delete head 被注释掉了，这是无锁数据结构最核心的难点。问题
线程 A 出队，CAS 成功把 head 从节点 X 移到 X->next，准备 delete X。但此时线程 B 可能刚读到旧的 head = X，正在访问 X->next。如果 A 先 delete 了 X，B 就访问了已释放内存（UAF）。
三种主流方案
表格
| 方案 | 原理 | 代表实现 | 优缺点 |
|---|---|---|---|
| Hazard Pointer | 线程声明 “我正在访问 X”，回收者检查所有 hazard pointer 后才释放 | Folly hazptr、libcds |
安全但扫描开销大 |
| Epoch-based Reclamation (EBR) | 分三个 epoch，当前 epoch 进入的线程都退出后才回收上上个 epoch 的垃圾 | Linux RCU、Crossbeam | 性能好，但回收延迟 |
| 乐观锁 + 延迟回收 | 用 tagged pointer 避免 ABA，节点放入延迟释放队列，定期批量回收 | Boost lockfree | 实现简单，内存占用高 |
简化版面试答案：可以说 ” 生产环境中通常用 Hazard Pointer 或 Epoch-based 回收，面试 demo 中可以用
std::atomic<Node*> + 版本号打包避免 ABA，节点用 GC 或延迟回收 “。五、SPSC 无锁环形缓冲区（更简单、更高性能）
如果是单生产者单消费者场景，环形缓冲区是最优解，不需要 CAS 循环，性能远超 Michael-Scott。
cpp
运行
template <typename T, size_t N>
class SPSCQueue {
private:
// 缓存行对齐，避免 false sharing
alignas(64) std::atomic<size_t> head_{0}; // 消费者写，生产者读
alignas(64) std::atomic<size_t> tail_{0}; // 生产者写，消费者读
T buffer_[N];
public:
bool enqueue(T value) {
size_t tail = tail_.load(std::memory_order_relaxed);
size_t next = (tail + 1) % N;
// 生产者只需要看 head 的最新值（acquire 保证看到消费者之前的写）
if (next == head_.load(std::memory_order_acquire))
return false; // 满
buffer_[tail] = std::move(value);
// release 保证 buffer 写入对消费者可见
tail_.store(next, std::memory_order_release);
return true;
}
bool dequeue(T& out) {
size_t head = head_.load(std::memory_order_relaxed);
if (head == tail_.load(std::memory_order_acquire))
return false; // 空
out = std::move(buffer_[head]);
head_.store((head + 1) % N, std::memory_order_release);
return true;
}
};
关键点：
head和tail分别在不同缓存行（alignas(64)），避免伪共享（False Sharing）- 生产者只写
tail、读head；消费者只写head、读tail—— 没有竞争，不需要 CAS - 内存序：写数据用
release，读对方指针用acquire
六、面试高频追问
Q1: CAS 失败重试会不会导致活锁（Livelock）？
Michael-Scott 队列是 lock-free 但不是 wait-free。极端情况下多个线程反复 CAS 失败可能短暂活锁，但有辅助推进机制（帮其他线程推进 tail）保证系统整体推进，实际不会永久活锁。
Q2: 为什么用 dummy node？
空队列时 head == tail == dummy。如果不用 dummy，入队第一个节点时需要同时修改 head 和 tail，两个原子操作无法原子地完成，会产生竞态。dummy node 让入队永远只改 tail->next 和 tail，出队永远只改 head，职责分离。
Q3: 无锁队列一定比有锁快吗？
不一定。
- 低竞争场景：有锁队列（
std::mutex+std::queue）因为不需要 CAS 重试和内存屏障，可能更快 - 高竞争场景：无锁队列优势明显，因为不会阻塞线程
- 内存回收开销可能抵消无锁收益
- 面试标准答案：取决于场景，SPSC 环形缓冲区 > 有锁 > MPMC 无锁链表（大致）
Q4: 如何实现有界无锁队列？
- 方案 1：环形缓冲区 + 原子索引（SPSC 简单，MPMC 需要 CAS）
- 方案 2：链表 + 原子计数器（enqueue 前 CAS 递增计数，满则拒绝）
Q5: C++ 中 compare_exchange_weak 和 strong 的区别？
weak可能伪失败（spurious failure）：值相等也返回 false，因为平台底层是 LL/SC（Load-Linked/Store-Conditional），性能略好strong保证不伪失败- 循环中用 weak（反正要重试），单次判断用 strong
七、总结：面试回答框架
被问到 “讲讲无锁队列” 时，按这个结构答：
- 一句话定义：用 CAS 等原子操作替代互斥锁的并发队列，保证系统级推进
- 核心原语：CAS + 内存序（acquire/release 配对）
- 经典算法：Michael-Scott 链表队列，讲入队 / 出队流程 + dummy node 作用
- 两大难题：ABA（tagged pointer）+ 安全内存回收（Hazard Pointer / Epoch）
- 性能优化：SPSC 环形缓冲区、缓存行对齐避免伪共享
- 适用场景：高并发、低延迟、不能阻塞的场景（网络包处理、任务调度、中断上下文）
这个回答覆盖了原理、实现、坑点、性能，面试足够深入。需要我针对某个点（比如 Hazard Pointer 具体实现、128 位 CAS、或者 Go/Java 中的无锁队列）再展开吗？