原子操作与内存序:并发编程的基石
说到多线程编程,锁是最常见的工具。但锁有个问题——它太重了。我在几年前优化一个高频交易系统时,发现锁竞争占了整个延迟的40%。那时候我就意识到,必须得用更轻量的东西。这就是我们今天要聊的原子操作。
std::atomic 基础
说白了,std::atomic 就是一个保证读写操作不会被线程调度打断的类型。它底层依赖CPU提供的原子指令,比如 x86 上的 LOCK 前缀。
核心要点:原子操作是不可分割的。要么全部执行完,要么一点都没执行。中间不会插入其他线程的操作。
来看个最简单的例子:
#include <atomic>
#include <thread>
#include <iostream>
std::atomic<int> counter(0);
void worker() {
for (int i = 0; i < 10000; ++i) {
counter.fetch_add(1); // 原子加1
}
}
int main() {
std::thread t1(worker);
std::thread t2(worker);
t1.join();
t2.join();
std::cout < "最终结果: " << counter << std::endl; // 一定是20000
return 0;
}
嗯,这里要注意。如果你用普通的 int 加 ++ 操作,两个线程同时执行时可能只加了一次。但用 fetch_add 就不会有这个问题。我在项目中见过太多因为没用原子操作导致的诡异bug,排查起来特别痛苦。
内存序(memory_order)
原子操作只是第一步。真正让C++原子库强大的,是内存序的控制。你想想看,CPU和编译器为了性能,会重排指令。单线程下没问题,多线程下就乱了。
C++标准定义了6种内存序:
| 内存序 | 含义 | 性能开销 |
|---|---|---|
| memory_order_relaxed | 只保证原子性,不保证顺序 | 最低 |
| memory_order_consume | 依赖关系排序(基本不用) | 较低 |
| memory_order_acquire | 后续读写不能重排到此操作之前 | 中等 |
| memory_order_release | 之前的读写不能重排到此操作之后 | 中等 |
| memory_order_acq_rel | acquire + release 的组合 | 较高 |
| memory_order_seq_cst | 全局顺序一致(默认) | 最高 |
我的建议:新手先用默认的 memory_order_seq_cst。等你能准确说出每个内存序的语义时,再考虑优化。我曾经为了省那点性能,用 relaxed 写出过一个只在特定CPU上复现的bug,教训深刻。
为什么会有性能差异?因为 seq_cst 需要在所有CPU核心之间同步,而 relaxed 只在当前核心保证原子性。在ARM架构上,这个差距尤其明显。
CAS操作:无锁编程的核心
CAS(Compare-And-Swap)是原子操作中最灵活的一个。它的逻辑是:如果当前值等于期望值,就更新为新值;否则什么都不做。整个过程是原子的。
std::atomic<int> value(0);
// CAS 的典型用法
int expected = 0;
int new_value = 42;
bool success = value.compare_exchange_weak(expected, new_value);
if (success) {
// value 原来是0,现在变成了42
} else {
// value 不是0,expected 被更新为当前值
}
这里有个坑:compare_exchange_weak 可能会因为伪失败(spurious failure)而返回false,即使值匹配。所以通常要放在循环里:
int expected = value.load();
while (!value.compare_exchange_weak(expected, new_value)) {
// expected 已经被更新,可以重试
}
而 compare_exchange_strong 不会有伪失败,但性能稍差。我个人习惯在循环里用 weak,在只尝试一次的场景用 strong。
无锁数据结构
有了CAS,我们就可以构建无锁数据结构了。无锁的意思是:至少有一个线程能继续推进,不会因为其他线程的阻塞而卡死。
来看一个最简单的无锁栈:
template<typename T>
class LockFreeStack {
private:
struct Node {
T data;
Node* next;
};
std::atomic<Node*> head{nullptr};
public:
void push(const T& value) {
Node* new_node = new Node{value, nullptr};
new_node->next = head.load();
while (!head.compare_exchange_weak(new_node->next, new_node)) {
// 如果 head 被其他线程改了,就重试
}
}
bool pop(T& result) {
Node* old_head = head.load();
while (old_head && !head.compare_exchange_weak(old_head, old_head->next)) {
// 重试直到成功或栈为空
}
if (old_head) {
result = old_head->data;
delete old_head; // 注意:这里可能有ABA问题
return true;
}
return false;
}
};
注意:上面的代码有ABA问题。线程1读取head为A,然后被挂起。线程2弹出A再压入B,然后又把B弹出压入A(地址相同但内容不同)。线程1恢复后,CAS发现head还是A,就认为没变化,但实际上栈结构已经变了。解决方法是使用带标记的指针(tagged pointer)或hazard pointer。
无锁数据结构的设计难点不在于CAS本身,而在于内存管理。你想想看,一个线程在读取节点时,另一个线程把它删了怎么办?这就是为什么工业级的无锁数据结构通常需要配合垃圾回收机制。
知识体系总览
下面这张图总结了本章的核心内容:
避坑指南
最后分享几个我踩过的坑:
- 不要用 volatile 做原子操作。 volatile 只防止编译器优化,不保证CPU级别的原子性。我曾经在一个嵌入式项目里看到有人用 volatile 做多线程同步,结果线上出了随机崩溃。
- 注意 false sharing。 多个原子变量如果在同一个缓存行,会导致性能急剧下降。用
alignas(64)来隔离它们。 - 无锁数据结构不是银弹。 在低竞争场景下,无锁结构确实比锁快。但高竞争时,CAS循环会导致总线风暴,性能反而不如精心设计的锁。
我的经验:在决定用无锁之前,先用性能分析工具看看锁竞争到底是不是瓶颈。很多时候,优化锁的粒度比换成无锁更有效。记住,最简单的方案往往是最好的方案。
原子操作和内存序是并发编程的底层工具。用好了,你能写出极致性能的代码。用不好,你会陷入难以调试的并发bug中。我建议你从简单的计数器开始练习,逐步过渡到无锁数据结构。多写、多测、多分析,慢慢就能掌握其中的门道。