为什么需要无锁编程
在多线程程序中，互斥锁（mutex）是最常用的同步原语，但它并非银弹。当线程竞争锁失败时，操作系统会将其挂起，引发一次开销可观的上下文切换。在高度竞争的场景下，多个线程反复获取和释放同一把锁，还会导致缓存行颠簸（cache line bouncing）——锁变量所在的缓存行在各个 CPU 核心之间来回迁移，严重损害性能。
无锁（lock-free）算法的目标是在不使用互斥锁的前提下，完成线程间的安全协作。它的优势主要体现在两个场景：
- 高并发竞争：无锁算法避免了上下文切换开销，线程始终处于运行态；
- 实时系统：无锁算法提供了更确定性的延迟上界，不会因为调度悬停而阻塞。
但需要清醒地认识到，无锁编程的复杂度远高于基于锁的编程。Dijkstra 曾警告我们：并发编程中最困难的不是保证正确性，而是说服自己（和他人）程序是正确的。无锁代码的每一条语句都必须经受内存模型的严格审视。
std::atomic 基础
C++11 引入的 std::atomic 模板是无锁编程的基石。它将普通的非原子类型包装成原子类型，保证对其的读写操作在硬件层面是不可分割的。
核心操作
#include <atomic>
#include <iostream>
int main() {
std::atomic<int> counter{0};
// 原子读取
int val = counter.load();
// 原子写入
counter.store(42);
// 原子交换：返回旧值，写入新值
int old = counter.exchange(100);
// CAS（Compare-And-Swap）：原子条件写入
int expected = 100;
bool success = counter.compare_exchange_strong(expected, 200);
// 若 counter == expected（100），则设为 200，返回 true
// 否则将当前值写入 expected，返回 false
}
compare_exchange_weak 与 compare_exchange_strong 的区别在于：weak 版本可能会伪失败（spurious failure），即在预期值与实际值相等时依然返回 false，这在某些架构上性能更优，通常用于循环重试场景中。
is_lock_free
并非所有平台上的 std::atomic<T> 都是无锁实现的。可以通过 is_lock_free() 查询：
std::atomic<long long> ll;
std::cout << ll.is_lock_free() << std::endl; // 1 表示无锁，0 表示内部使用了锁
指针与平凡类型
std::atomic 可以包装指针和满足平凡可复制（trivially copyable）条件的结构体。对于自定义类型，编译器会检查其是否可以被原子操作安全处理。
std::atomic_flag
C++11 标准中唯一保证无锁的原子类型是 std::atomic_flag。它只有两个状态（set / clear），接口极简，常被用作实现自旋锁的底层构件：
#include <atomic>
class Spinlock {
std::atomic_flag flag = ATOMIC_FLAG_INIT;
public:
void lock() {
while (flag.test_and_set(std::memory_order_acquire)) {
// 自旋等待
}
}
void unlock() {
flag.clear(std::memory_order_release);
}
};
这里 test_and_set 使用 memory_order_acquire，clear 使用 memory_order_release，构成了一对经典的同步关系。
内存序模型详解
原子操作保证操作本身的不可分割性，但内存序决定了操作结果对其他线程的可见顺序。C++11 提供了六种内存序（memory ordering），从弱到强排列：
memory_order_relaxed
纯原子性，无任何跨线程的顺序保证。编译器和处理器可以任意重排 relaxed 操作。
std::atomic<int> x{0};
// Thread 1
x.store(1, std::memory_order_relaxed);
// Thread 2
int v = x.load(std::memory_order_relaxed); // v 可能读到 1，也可能读到 0，取决于时序
Relaxed 适用于只需要原子计数器、不要求顺序语义的场景，例如简单的引用计数。
memory_order_acquire
Acquire 语义用于读操作。它保证：所有在该 acquire load 之后执行的内存访问，不会被重排到该 load 之前。换句话说，acquire 像一个单向闸门，允许前面的操作透过来，但阻止后面的操作溜到前面。
memory_order_release
Release 语义用于写操作。它保证：所有在该 release store 之前执行的内存访问，不会被重排到该 store 之后。
memory_order_acq_rel
同时具有 acquire 和 release 语义，用于读-改-写（RMW）操作，如 fetch_add 或 CAS。
memory_order_seq_cst
顺序一致性（Sequential Consistency）是默认也是最强的内存序。它在 acquire/release 的基础上，额外保证所有线程以相同的顺序观察到所有 seq_cst 操作。这极大地简化了推理逻辑，但也是性能开销最大的选择。
指令重排与 happens-before
现代 CPU 和编译器为了性能，会重新排序指令。内存序的本质就是在这两者之间划定边界，防止某些重排破坏程序语义。
Release-Acquire 对构成了 C++ 内存模型中最核心的同步机制：
std::atomic<int> ready{0};
int data = 0;
// Thread 1（生产者）
data = 42; // A
ready.store(1, std::memory_order_release); // B
// Thread 2（消费者）
while (ready.load(std::memory_order_acquire) != 1) { // C
// 等待
}
std::cout << data; // D
如果 Thread 2 在 C 点读到 ready 为 1，那么根据 synchronizes-with 关系，B 点之前的写操作（A）对 Thread 2 可见。于是 D 一定输出 42。这种“发布/订阅”模式是无锁编程中最常见的同步范式。
Happens-Before 与 Synchronizes-With
Happens-before 是 C++ 内存模型中定义程序正确性的核心概念。如果操作 A happens-before 操作 B，那么 A 的效果对 B 可见。
Synchronizes-with 是 happens-before 的一种特殊跨线程形式。当一个线程执行了 release store，另一个线程随后执行了 acquire load 并读到了该值，这两个操作之间就建立了 synchronizes-with 关系。
一个常见的误区是认为 seq_cst 是唯一"安全"的选择。事实上，在大多数无锁算法中，release/acquire 已经足够。seq_cst 只在以下少数场景不可或缺：
- 需要多变量间的全序关系（例如 Dekker 算法）；
- 安全地实现双标志锁（flag-based locking）。
对于普通的数据结构实现，显式使用 release/acquire 不仅正确，还能避免 seq_cst 带来的额外内存栅栏开销。
ABA 问题
ABA 问题是无锁编程中最隐蔽的陷阱之一。考虑一个使用 CAS 实现的锁-free 栈：
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, head.load(std::memory_order_relaxed)};
while (!head.compare_exchange_weak(
new_node->next, new_node,
std::memory_order_release,
std::memory_order_relaxed));
}
std::shared_ptr<T> pop() {
Node* old_head = head.load(std::memory_order_acquire);
while (old_head && !head.compare_exchange_weak(
old_head, old_head->next,
std::memory_order_release,
std::memory_order_relaxed)) {
}
// ... 危险：old_head 可能已被其他线程 delete 并重新 new 出来
}
};
问题在于：线程 A 读取 head 得到指针 P，随后线程 B 将 P 弹出并释放内存，接着线程 B（或另一个线程 C）又 push 了一个新节点，恰好操作系统分配到了同样的地址 P。此时线程 A 的 CAS 成功，但栈的实际状态已完全不同——这就是 ABA。
ABA 的解决方案
Tagged Pointer（标记指针）：将指针与其修改次数打包在一个 64 位或 128 位整数中，每次修改递增计数器。C++ 的
std::atomic<std::uintptr_t>可以用于此目的（需要保证足够位宽）。延迟回收（Hazard Pointers / Epoch-Based Reclamation）：pop 操作不立即 delete 节点，而是将其放入一个待回收列表，等待所有可能访问该节点的线程都离开临界区后再安全释放。
RCU（Read-Copy-Update）：读取者无锁访问，写入者复制一份数据修改后原子切换指针，旧数据延迟释放。
无锁数据结构实现
无锁栈（完整实现）
下面的实现使用了一个简单的垃圾回收策略——所有弹出的节点放入一个 retire list，在程序退出或特定时机统一释放。工业级代码应使用 hazard pointers，但此实现足以展示核心逻辑：
#include <atomic>
#include <memory>
#include <vector>
template<typename T>
class LockFreeStack {
struct Node {
std::shared_ptr<T> data;
Node* next;
Node(T const& value)
: data(std::make_shared<T>(value)), next(nullptr) {}
};
std::atomic<Node*> head;
std::atomic<int> threads_in_pop{0};
std::atomic<Node*> to_be_deleted{nullptr};
static void delete_nodes(Node* nodes) {
while (nodes) {
Node* next = nodes->next;
delete nodes;
nodes = next;
}
}
void chain_pending_nodes(Node* nodes) {
Node* last = nodes;
while (Node* const next = last->next) {
last = next;
}
chain_pending_nodes(nodes, last);
}
void chain_pending_nodes(Node* first, Node* last) {
last->next = to_be_deleted.load(std::memory_order_relaxed);
while (!to_be_deleted.compare_exchange_weak(
last->next, first,
std::memory_order_release,
std::memory_order_relaxed));
}
void chain_pending_node(Node* n) {
chain_pending_nodes(n, n);
}
void try_reclaim(Node* old_head) {
if (threads_in_pop.load(std::memory_order_acquire) == 1) {
Node* nodes_to_delete = to_be_deleted.exchange(nullptr, std::memory_order_acquire);
if (!--threads_in_pop) {
delete_nodes(nodes_to_delete);
} else if (nodes_to_delete) {
chain_pending_nodes(nodes_to_delete);
}
delete old_head;
} else {
chain_pending_node(old_head);
--threads_in_pop;
}
}
public:
LockFreeStack() : head(nullptr) {}
void push(T const& value) {
Node* const new_node = new Node(value);
new_node->next = head.load(std::memory_order_relaxed);
while (!head.compare_exchange_weak(
new_node->next, new_node,
std::memory_order_release,
std::memory_order_relaxed));
}
std::shared_ptr<T> pop() {
++threads_in_pop;
Node* old_head = head.load(std::memory_order_acquire);
while (old_head && !head.compare_exchange_weak(
old_head, old_head->next,
std::memory_order_release,
std::memory_order_acquire)) {
}
std::shared_ptr<T> res;
if (old_head) {
res.swap(old_head->data);
}
try_reclaim(old_head);
return res;
}
~LockFreeStack() {
delete_nodes(head.load());
delete_nodes(to_be_deleted.load());
}
};
关键点：
push使用 release 语义将新节点发布出去；pop使用 acquire 语义读取 head，保证看到完整的已发布节点；- CAS 的 failure 路径使用 acquire，保证重试时读到最新的 head。
简化版无锁队列（Michael-Scott Queue）
Michael-Scott 队列是无锁队列的经典实现。其核心思想是维护 head 和 tail 两个原子指针，enqueue 时在尾部 CAS 挂载新节点，dequeue 时在头部 CAS 移出节点：
#include <atomic>
#include <memory>
template<typename T>
class LockFreeQueue {
struct Node {
std::shared_ptr<T> data;
std::atomic<Node*> next;
Node() : next(nullptr) {}
};
std::atomic<Node*> head;
std::atomic<Node*> tail;
// 简化版：省略了 hazard pointer 和 ABA 防护
public:
LockFreeQueue() {
Node* dummy = new Node();
head.store(dummy, std::memory_order_relaxed);
tail.store(dummy, std::memory_order_relaxed);
}
void enqueue(T value) {
std::shared_ptr<T> new_data = std::make_shared<T>(std::move(value));
Node* p = new Node();
p->data = new_data;
Node* old_tail = tail.load(std::memory_order_acquire);
for (;;) {
Node* null_ptr = nullptr;
if (old_tail->next.compare_exchange_strong(
null_ptr, p,
std::memory_order_release,
std::memory_order_relaxed)) {
// 成功挂载到尾部
tail.compare_exchange_strong(old_tail, p,
std::memory_order_release,
std::memory_order_relaxed);
break;
} else {
// 尾部已前进，帮助推进 tail
tail.compare_exchange_strong(old_tail,
old_tail->next.load(std::memory_order_acquire),
std::memory_order_release,
std::memory_order_relaxed);
old_tail = tail.load(std::memory_order_acquire);
}
}
}
std::shared_ptr<T> dequeue() {
Node* old_head = head.load(std::memory_order_acquire);
for (;;) {
Node* old_tail = tail.load(std::memory_order_acquire);
Node* old_head_next = old_head->next.load(std::memory_order_acquire);
if (old_head == old_tail) {
if (!old_head_next) {
return std::shared_ptr<T>(); // 空队列
}
// tail 滞后，帮助推进
tail.compare_exchange_strong(old_tail, old_head_next,
std::memory_order_release,
std::memory_order_relaxed);
} else {
std::shared_ptr<T> res = old_head_next->data;
if (head.compare_exchange_strong(old_head, old_head_next,
std::memory_order_release,
std::memory_order_relaxed)) {
// 释放旧 dummy 节点的责任交给调用者或 GC
return res;
}
}
}
}
};
注意：这是一个教学简化版，没有处理节点的安全释放问题。生产环境应集成 hazard pointers 或 epoch-based reclamation。
内存栅栏（Memory Fences）
有时需要在没有数据依赖的独立原子变量之间建立同步关系，这时可以使用显式内存栅栏：
std::atomic_thread_fence(std::memory_order_release);
std::atomic_thread_fence(std::memory_order_acquire);
std::atomic_thread_fence(std::memory_order_seq_cst);
栅栏与原子操作的语义区别在于：栅栏不操作任何特定变量，而是对所有内存访问施加顺序约束。
编译器栅栏 vs 硬件栅栏
编译器为了不破坏单线程语义，会避免某些重排，但跨线程视角下的重排是允许的。std::atomic_thread_fence 会同时向编译器和 CPU 发出指令，阻止特定方向的重排。
例如，memory_order_release 的栅栏保证：栅栏之前的所有内存写操作都完成后，栅栏之后的写操作才能开始。memory_order_acquire 的栅栏则保证：栅栏之后的读操作不会重排到栅栏之前。
何时使用显式栅栏
显式栅栏很少见，通常以下场景可能需要它们：
- 需要同步一个释放操作与多个不相关的获取操作；
- 需要以非原子变量为中介建立 happens-before 关系；
- 实现极致优化的底层库代码。
对绝大多数应用开发者而言，正确地选择 std::atomic 操作上的内存序参数就够了。
总结
C++11 内存模型为多线程编程提供了严格而精确的理论基础。理解 memory ordering 的关键不在于记住重排规则表，而在于建立直觉：release 是"后置承诺"，acquire 是"前置等待"，它们之间的交汇点就是 synchronizes-with。
使用无锁算法时，应遵循以下原则：
- 默认使用 mutex：除非有明确的性能瓶颈和测量数据支持，不要追求无锁；
- 显式指定 memory_order：不要依赖默认的 seq_cst，它可能隐藏性能问题；
- 警惕 ABA：所有基于 CAS 的循环链表结构都必须考虑回收安全；
- 用 sanitizer 和 stress test 验证：无锁 bug 的出现概率低但破坏力大，常规测试难以覆盖。
无锁编程是 C++ 并发领域的深水区。掌握 std::atomic 和内存序只是第一步，真正的挑战在于设计能够正确推理、经得起硬件重排考验的算法。在动手实现之前，务必确认现有库（如 Boost.Lockfree）是否已能满足需求——站在巨人的肩膀上，总比自己重新发明轮子更安全。
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。