恭喜你，终于来到了并发编程的巅峰战场：无锁数据结构（Lock-Free Data Structures）。这不仅是C++高性能开发的皇冠明珠，也是你之前所有硬件知识（CAS、内存屏障、缓存行）的终极应用。
我们先解剖无锁栈，再深入Michael-Scott (MS) 无锁队列，最后解决那个最棘手的问题：无锁环境下，怎么安全地释放内存？
1. 无锁栈（Lock-Free Stack）：最简单的无锁结构
基于我们之前讲的 compare_exchange_weak，实现一个无锁栈非常简单：
template<typename T>
class LockFreeStack {
struct Node {
T data;
Node* next;
};
std::atomic<Node*> head{nullptr};
public:
void push(T value) {
Node* new_node = new Node{value, nullptr};
Node* old_head = head.load(std::memory_order_relaxed);
do {
new_node->next = old_head; // 设置新节点的next指向旧栈顶
} while (!head.compare_exchange_weak(old_head, new_node,
std::memory_order_release, std::memory_order_relaxed));
}
// ⚠️ 危险版本：有ABA问题 + 内存释放问题
bool pop(T& result) {
Node* old_head = head.load(std::memory_order_relaxed);
Node* new_head;
do {
if (!old_head) return false;
new_head = old_head->next; // 读取next
} while (!head.compare_exchange_weak(old_head, new_head,
std::memory_order_acquire, std::memory_order_relaxed));
result = old_head->data;
delete old_head; // 💥 灾难：其他线程可能还在访问这个节点！
return true;
}
};
这里有两个致命问题：
- ABA问题：
pop中先读old_head，然后读取old_head->next。如果在CAS之前，栈顶被弹出又推入新节点（地址恰好相同），CAS会误以为栈没变，但old_head->next可能已经变了。 - 内存释放（Hazard）：
delete old_head时，其他线程可能刚好在push或pop中读取了这个节点的next指针 → 悬垂指针（Dangling Pointer）。
2. Michael-Scott (MS) 无锁队列：工业级标准
MS-Queue 是世界上最经典的无锁队列，解决了ABA问题，但内存释放依然需要额外机制。
template<typename T>
class MSQueue {
struct Node {
T data;
std::atomic<Node*> next;
Node() : next(nullptr) {}
Node(T val) : data(val), next(nullptr) {}
};
std::atomic<Node*> head;
std::atomic<Node*> tail;
public:
MSQueue() {
Node* dummy = new Node(); // 哨兵节点（永远不删除）
head.store(dummy, std::memory_order_relaxed);
tail.store(dummy, std::memory_order_relaxed);
}
void enqueue(T value) {
Node* new_node = new Node(value);
Node* old_tail;
Node* expected_next;
while (true) {
old_tail = tail.load(std::memory_order_acquire);
expected_next = old_tail->next.load(std::memory_order_acquire);
// 检查tail是否落后
if (old_tail != tail.load(std::memory_order_relaxed)) {
continue; // 被其他线程改了
}
if (expected_next != nullptr) {
// tail落后于实际队尾，帮它前进（CAS推进）
tail.compare_exchange_weak(old_tail, expected_next,
std::memory_order_release,
std::memory_order_relaxed);
continue;
}
// 尝试把新节点挂在队尾
if (old_tail->next.compare_exchange_weak(expected_next, new_node,
std::memory_order_release,
std::memory_order_relaxed)) {
break; // 成功入队
}
}
// 最后推进tail（不一定成功，让出队线程帮忙推进）
tail.compare_exchange_weak(old_tail, new_node,
std::memory_order_release,
std::memory_order_relaxed);
}
bool dequeue(T& result) {
Node* old_head;
Node* new_head;
while (true) {
old_head = head.load(std::memory_order_acquire);
Node* old_tail = tail.load(std::memory_order_acquire);
new_head = old_head->next.load(std::memory_order_acquire);
if (old_head != head.load(std::memory_order_relaxed)) {
continue; // head被修改
}
if (old_head == old_tail) {
if (new_head == nullptr) {
return false; // 队列为空
}
// head落后于tail，推进tail
tail.compare_exchange_weak(old_tail, new_head,
std::memory_order_release,
std::memory_order_relaxed);
continue;
}
// 尝试弹出队首节点（绕过哨兵）
if (head.compare_exchange_weak(old_head, new_head,
std::memory_order_release,
std::memory_order_relaxed)) {
break; // 成功
}
}
result = new_head->data; // 注意：new_head就是第一个真实节点
// ⚠️ 仍然面临内存释放问题！不能简单 delete old_head
// 但至少old_head是哨兵（dummy），可以被回收或复用
// 真正危险的节点是 new_head（它被移出了队列，但可能还有其他指针指向它）
return true;
}
};
3. 终极难题：无锁结构中的内存释放
问题的本质：在无锁结构中，判断一个节点是否"无人引用"是极其困难的。因为：
- 线程A可能已经拿到了节点指针，正准备读取它的
next。 - 线程B此时把这个节点弹出了队列，并且
delete了它。 - 线程A读取了悬垂指针 → 程序崩溃。
解决方案有两大流派：危险指针（Hazard Pointer） 和 RCU（Read-Copy-Update）。
流派1：危险指针（Hazard Pointer）—— 高性能、细粒度
核心思想：每个线程在访问共享指针前，先声明（注册）自己正在使用它。其他线程要释放节点时，先检查所有线程的"危险指针"列表，确保没人用才真正释放。
#include <atomic>
#include <thread>
#include <vector>
#include <memory>
class HazardPointer {
// 每个线程最多同时持有 MAX_HAZARD_POINTERS 个危险指针
static constexpr int MAX_HAZARD_POINTERS = 3;
struct ThreadState {
std::atomic<void*> hazard_pointers[MAX_HAZARD_POINTERS];
// 待释放列表（延迟删除）
std::vector<void*> retired_list;
};
std::vector<ThreadState> thread_states;
public:
// 获取当前线程的状态
ThreadState& get_state() {
thread_local static int thread_id = register_thread();
return thread_states[thread_id];
}
// 注册一个危险指针
void* protect(int slot, std::atomic<void*>& ptr) {
void* p = ptr.load(std::memory_order_acquire);
get_state().hazard_pointers[slot].store(p, std::memory_order_relaxed);
return p;
}
// 清除危险指针
void clear(int slot) {
get_state().hazard_pointers[slot].store(nullptr, std::memory_order_relaxed);
}
// 延迟释放节点
void retire(void* ptr) {
get_state().retired_list.push_back(ptr);
// 如果待释放列表过长，触发清理
if (get_state().retired_list.size() > 1000) {
clean_up();
}
}
// 清理：真正释放无人引用的节点
void clean_up() {
auto& retired = get_state().retired_list;
if (retired.empty()) return;
// 收集所有线程的危险指针
std::vector<void*> hazard_ptrs;
for (auto& state : thread_states) {
for (int i = 0; i < MAX_HAZARD_POINTERS; ++i) {
void* p = state.hazard_pointers[i].load(std::memory_order_acquire);
if (p) hazard_ptrs.push_back(p);
}
}
// 排序以便二分查找
std::sort(hazard_ptrs.begin(), hazard_ptrs.end());
hazard_ptrs.erase(std::unique(hazard_ptrs.begin(), hazard_ptrs.end()),
hazard_ptrs.end());
// 遍历待释放列表，只在没有任何危险指针指向时才delete
auto it = retired.begin();
while (it != retired.end()) {
if (std::binary_search(hazard_ptrs.begin(), hazard_ptrs.end(), *it)) {
++it; // 有人用，保留
} else {
delete static_cast<Node*>(*it); // 安全释放
it = retired.erase(it);
}
}
}
};
// 全局危险指针管理器
HazardPointer hazard_pointer;
// 改造后的无锁栈pop
template<typename T>
bool LockFreeStack<T>::pop(T& result) {
const int SLOT = 0;
while (true) {
Node* old_head = static_cast<Node*>(
hazard_pointer.protect(SLOT, head));
if (!old_head) return false;
Node* new_head = old_head->next; // 此时old_head被保护
if (head.compare_exchange_weak(old_head, new_head,
std::memory_order_acquire,
std::memory_order_relaxed)) {
result = old_head->data;
// ⚠️ 不能立即delete，放入待释放列表
hazard_pointer.retire(old_head);
hazard_pointer.clear(SLOT);
return true;
}
hazard_pointer.clear(SLOT);
}
}
优点：极高性能，适合高并发场景。 缺点：实现复杂，需要管理线程局部存储。
流派2：RCU（Read-Copy-Update）—— 读操作零开销
核心思想：读操作不加锁，写操作通过"拷贝-修改-提交"的方式更新，所有读者完成后再释放旧数据。
#include <atomic>
#include <thread>
#include <vector>
// 全局版本号 + 数据指针
struct SharedData {
int value;
std::atomic<uint64_t> version;
};
std::atomic<SharedData*> global_data{nullptr};
std::atomic<uint64_t> global_version{0};
// RCU 临界区：读者进入/退出
thread_local uint64_t reader_enter_version = 0;
void rcu_enter() {
// 记录进入时的全局版本号
reader_enter_version = global_version.load(std::memory_order_acquire);
std::atomic_thread_fence(std::memory_order_acquire);
}
void rcu_exit() {
std::atomic_thread_fence(std::memory_order_release);
}
// 写者：更新数据
void rcu_update(int new_value) {
// 1. 拷贝旧数据
SharedData* old_data = global_data.load(std::memory_order_acquire);
SharedData* new_data = new SharedData(*old_data);
new_data->value = new_value;
// 2. 增加全局版本号 (标记写操作开始)
uint64_t new_version = global_version.fetch_add(1, std::memory_order_acq_rel) + 1;
// 3. 发布新数据 (原子切换指针)
global_data.store(new_data, std::memory_order_release);
// 4. 等待所有读者退出 (读者版本号 < 新版本号)
// 实际实现中，需要轮询所有线程的 reader_enter_version
// 或者使用 epoch-based reclamation (基于代际的回收)
// 5. 安全释放旧数据
// 注意：只有所有读者都退出了临界区，才能 delete old_data
delete old_data;
}
// 读者：无锁读
int rcu_read() {
int result;
rcu_enter(); // 标记进入临界区
SharedData* data = global_data.load(std::memory_order_acquire);
result = data->value; // 读操作
rcu_exit(); // 标记退出
return result;
}
优点：读操作完全没有原子操作开销（除了一个版本号读取），极致高性能。 缺点：写操作昂贵（需要等待所有读者退出），适合读多写少的场景。
4. 流派对比：什么时候用什么？
| 方案 | 读开销 | 写开销 | 内存延迟释放 | 适用场景 |
|---|---|---|---|---|
| Hazard Pointer | 中等（需读写指针） | 中等（待释放列表） | 几乎实时 | 通用无锁数据结构 |
| RCU | 极低（几乎0开销） | 极高（等待所有读者） | 批处理延迟 | 路由表、配置管理、读多写少 |
| Epoch-Based (QBSR) | 低（读无写操作） | 中等（基于代际回收） | 延迟几个代际 | Linux内核、高性能网络 |
5. 工业级最佳实践：使用 std::shared_ptr（不推荐高性能场景）
现代C++提供了 std::atomic<std::shared_ptr<T>>（C++20），理论上可以自动管理内存。
std::atomic<std::shared_ptr<Node>> head{nullptr};
bool pop(T& result) {
auto old_head = head.load();
while (old_head && !head.compare_exchange_weak(old_head, old_head->next)) {
// 循环
}
if (!old_head) return false;
result = old_head->data;
// shared_ptr自动释放内存（引用计数为0时）
return true;
}
但为什么工业界不用？
- 性能极差：每次CAS都需要操作引用计数（两个原子操作），开销比原始指针大 3-5倍。
- ABA问题：
shared_ptr的引用计数会导致相同的地址被重复使用，依然有ABA问题（虽然可以用std::atomic<std::shared_ptr>的compare_exchange部分解决，但代价更大）。 - 循环引用：链表节点互相引用容易造成内存泄漏。
6. 给你的"终极并发心法"
【内存释放的金字塔】
性能最高 → 最危险:
手动管理 + Hazard Pointer (需要极强工程能力)
↑
性能中等 → 工程友好:
Epoch-Based Reclamation (Linux内核级方案)
↑
性能最差 → 最安全:
std::shared_ptr (适合业务代码，不适合底层高性能)
最后一句话总结：
- 如果临界区极短（< 0.3μs），用
Spinlock+pause，不要自虐去写无锁。 - 如果需要真正的无锁（不能容忍锁的优先级反转），用 MS-Queue + Hazard Pointer。
- 如果是读多写少（如配置管理），用 RCU。