谈谈 lock-free queue 如何实现？
Lock-free 编程是 C++ 并发里最难的领域，NVIDIA 面试会深挖这块。从原理到实现系统讲一遍。
为什么需要 Lock-Free
Mutex 的几个核心问题：
优先级反转：低优先级线程持有锁，高优先级线程被迫等待，实时系统（游戏 tick、音频线程）无法接受。
Convoying（护送效应）：一个线程持锁慢，所有竞争线程都在 OS 队列里排队，吞吐量崩溃。
Cache line 乒乓：mutex 内部的状态字节被多个核反复争夺，cache 一致性协议产生大量总线流量。
Lock-free 的定义：系统整体总有线程在推进，单个线程可能重试，但不会有线程被永久阻塞。
核心原语：CAS 与 Memory Order
所有 lock-free 结构都建立在 Compare-And-Swap 上：
cpp
// 伪语义（CPU 保证这是原子操作）
bool CAS(T* addr, T expected, T desired) {
if (*addr == expected) {
*addr = desired;
return true;
}
return false;
}
// C++ 中
std::atomic<int*> ptr;
int* expected = old_ptr;
// weak 版本：允许偶发性失败（spurious failure），用在循环里性能更好
ptr.compare_exchange_weak(expected, new_ptr,
std::memory_order_release, // 成功时的 ordering
std::memory_order_relaxed); // 失败时的 ordering
Memory Order 速查（这是面试必考点）：
relaxed — 只保证原子性，不保证顺序。用于计数器、不需要同步的读 acquire — 此操作之后的读写不能被重排到此之前。用于"读锁" release — 此操作之前的读写不能被重排到此之后。用于"写锁" seq_cst — 全局顺序一致，最安全但最慢，是默认值 经典配对：producer 用 release 写，consumer 用 acquire 读
第一步：SPSC Queue（最实用）
单生产者单消费者队列，游戏引擎里最常见：音频线程与游戏线程通信、渲染命令缓冲区。SPSC 不需要 CAS，只用 acquire/release 就够了。
cpp
template<typename T, size_t N>
class SPSCQueue {
// N 必须是 2 的幂，用位运算代替取模
static_assert((N & (N - 1)) == 0, "N must be power of 2");
static constexpr size_t MASK = N - 1;
// 关键：head 和 tail 必须在不同 cache line！
// 否则 producer 和 consumer 会反复争夺同一 cache line（false sharing）
alignas(64) std::atomic<size_t> head_{0}; // consumer 更新
alignas(64) std::atomic<size_t> tail_{0}; // producer 更新
T buffer_[N];
public:
// Producer 调用（只有一个线程调用）
bool push(const T& item) {
size_t tail = tail_.load(std::memory_order_relaxed); // 读自己的 tail，relaxed 够了
size_t next = (tail + 1) & MASK;
// 检查队列是否满：需要看 consumer 的 head，用 acquire 同步
if (next == head_.load(std::memory_order_acquire))
return false; // full
buffer_[tail] = item;
// 发布 tail，用 release：保证上面的写入对 consumer 可见
tail_.store(next, std::memory_order_release);
return true;
}
// Consumer 调用（只有一个线程调用）
bool pop(T& item) {
size_t head = head_.load(std::memory_order_relaxed); // 读自己的 head
// 检查队列是否空：需要看 producer 的 tail，用 acquire 同步
if (head == tail_.load(std::memory_order_acquire))
return false; // empty
item = buffer_[head];
// 发布 head，用 release：让 producer 知道这个槽位已经被消费
head_.store((head + 1) & MASK, std::memory_order_release);
return true;
}
size_t size() const {
size_t tail = tail_.load(std::memory_order_acquire);
size_t head = head_.load(std::memory_order_acquire);
return (tail - head + N) & MASK;
}
};
// 使用示例：音频线程 ↔ 游戏线程
SPSCQueue<AudioCommand, 1024> audio_queue;
// 游戏线程（producer）
audio_queue.push(AudioCommand{.type = PLAY, .sound_id = 42});
// 音频线程（consumer）
AudioCommand cmd;
if (audio_queue.pop(cmd)) {
play_sound(cmd.sound_id);
}
第二步：Michael-Scott Queue（MPMC 经典算法）
多生产者多消费者，1996 年论文，是绝大多数无锁队列的基础。用带哨兵节点的链表，CAS 竞争队尾/队头。
cpp
template<typename T>
class MSQueue {
struct Node {
T data;
std::atomic<Node*> next{nullptr};
Node() = default;
explicit Node(T val) : data(std::move(val)) {}
};
alignas(64) std::atomic<Node*> head_; // 指向哨兵节点
alignas(64) std::atomic<Node*> tail_; // 指向最后一个节点
public:
MSQueue() {
Node* dummy = new Node(); // 哨兵节点：head 永远指向一个已消费/空节点
head_.store(dummy, std::memory_order_relaxed);
tail_.store(dummy, std::memory_order_relaxed);
}
void push(T val) {
Node* node = new Node(std::move(val));
while (true) {
Node* tail = tail_.load(std::memory_order_acquire);
Node* next = tail->next.load(std::memory_order_acquire);
// 二次确认 tail 没变（其他线程可能推进了 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（失败也没关系，下个 push 会帮忙）
tail_.compare_exchange_weak(
tail, node,
std::memory_order_release,
std::memory_order_relaxed);
return;
}
} else {
// tail 落后了（另一个线程已经插入节点但还没推进 tail）
// 帮它推进 tail，再重试
tail_.compare_exchange_weak(
tail, next,
std::memory_order_release,
std::memory_order_relaxed);
}
}
}
bool pop(T& val) {
while (true) {
Node* 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_weak(tail, next,
std::memory_order_release, std::memory_order_relaxed);
} else {
// 读数据（在 CAS 之前！CAS 成功后 head 可能被别人 delete）
val = next->data;
if (head_.compare_exchange_weak(head, next,
std::memory_order_release,
std::memory_order_relaxed)) {
delete head; // ← 这里有 ABA 问题！下面专门讲
return true;
}
}
}
}
};
ABA 问题：Lock-Free 最大的坑
这是面试最高频的追问，必须说清楚。
场景：
初始状态：head → A → B → C 线程 1：读到 head = A，准备 CAS(head, A, B)，被挂起 线程 2：pop A，pop B，push A（内存分配器复用了 A 的地址） 现在：head → A' → C （A' 和 A 地址一样，但内容变了） 线程 1 恢复：CAS(head, A, B) A 的地址没变，CAS 成功！ 但 B 已经不在队列里了，head 指向了一个野指针
解法一：带标记的指针（Tagged Pointer）
利用指针对齐后低位恒为 0 的特性，塞一个版本号：
cpp
struct TaggedPtr {
Node* ptr;
uint64_t tag; // 每次 CAS 成功就 +1
};
// 用 128-bit CAS（x86 上是 CMPXCHG16B）
std::atomic<TaggedPtr> head_;
// A 被复用时，tag 已经是 2 了，不等于线程 1 保存的 tag=0
// CAS 失败，重试
解法二：Hazard Pointer（工业级方案）
cpp
// 每个线程声明"我正在访问这个指针，不要释放它"
thread_local Node* hazard_ptr = nullptr;
bool pop(T& val) {
while (true) {
Node* head = head_.load(std::memory_order_acquire);
hazard_ptr = head; // 声明保护
// 再次确认 head 没变（声明保护之后再验证）
if (head != head_.load(std::memory_order_acquire)) continue;
// ... 执行 pop 逻辑 ...
if (head_.compare_exchange_weak(...)) {
hazard_ptr = nullptr;
// 不立刻 delete，先放入"退休列表"
// 扫描所有线程的 hazard_ptr，确认没人持有再释放
retire(head);
return true;
}
}
}
解法三：Epoch-Based Reclamation（最常用于游戏/高性能场景）
全局 epoch：0 / 1 / 2 轮转 线程进入临界区：记录当前 epoch 退出临界区：清除记录 回收内存：确认所有线程都离开了某个 epoch 之后，批量释放
这是 folly（Facebook）、jemalloc 等库的实际做法，开销最低。
False Sharing：容易忽视的性能杀手
cpp
// 错误：head 和 tail 在同一 cache line（通常 64 字节）
struct BadQueue {
std::atomic<size_t> head; // consumer 写
std::atomic<size_t> tail; // producer 写
// 两个核各自频繁写这两个变量
// cache 一致性协议让这个 cache line 在两个核之间不断 bounce
// 性能比加锁还差！
};
// 正确：强制对齐到不同 cache line
struct GoodQueue {
alignas(64) std::atomic<size_t> head;
alignas(64) std::atomic<size_t> tail;
};
实测数据：在高竞争场景下，加 alignas(64) 可以让吞吐量提升 3-5 倍。
游戏里的实际应用
| 场景 | 队列类型 | 原因 |
|---|---|---|
| 音频命令（游戏线程 → 音频线程） | SPSC Ring Buffer | 单生产者单消费者，最快 |
| Job System 任务提交 | MPMC | 多个线程提交任务 |
| Work Stealing（Naughty Dog 的 Fiber 系统） | 每线程一个 SPSC + 偷取用 MPSC | 局部性好 + 负载均衡 |
| 网络包接收队列 | MPSC（多 IO 线程 → 单逻辑线程） | 多网卡 IO 线程并发写 |
| 渲染命令缓冲 | SPSC | 游戏线程生产，渲染线程消费 |
面试时的答题思路
NVIDIA 面试官问这道题，实际在考三件事：你知不知道 CAS 和 memory order 的配合；你有没有踩过或者理解 ABA 问题；你知不知道 false sharing 这个隐形性能杀手。
回答建议：先说 SPSC（最简单、最实用，配合 acquire/release 解释 memory order）；再说 MS Queue（CAS 竞争机制）；然后主动提 ABA 问题，说出至少一种解法。最后加一句 alignas(64) 防 false sharing，展示你真的写过这类代码。
如果被追问”你实际用过哪种”，把你项目里的真实场景说出来，比泛泛描述算法有说服力得多。