12. 原子操作与内存序：std::atomic 的使用、memory_order 详解、避免 ABA 问题

多线程编程里，最让人头疼的就是数据竞争。你辛辛苦苦写了个全局计数器，两个线程同时加一，结果跑出来比预期少了好几次。嗯，这就是典型的「非原子操作」带来的问题。

今天我们来聊聊 C++ 里解决这类问题的核心工具——std::atomic。我做了这么多年性能优化，可以负责任地说：原子操作用得好，并发问题少一半。

12.1 std::atomic 的基本用法

先看一个最简单的例子。假设我们有个共享计数器：

#include <atomic>
#include <thread>
#include <iostream>

std::atomic<int> counter(0);

void worker() {
    for (int i = 0; i < 10000; ++i) {
        counter.fetch_add(1);  // 原子加一
    }
}

int main() {
    std::thread t1(worker);
    std::thread t2(worker);
    t1.join();
    t2.join();
    std::cout << counter.load() << std::endl;  // 输出 20000
    return 0;
}

如果你用普通的 int 加 ++，结果大概率不是 20000。为什么？因为 counter++ 在底层是「读-改-写」三步，两个线程可能同时读到同一个值，然后各自加一写回去——结果只加了一次。

std::atomic 保证这些操作是不可分割的。说白了，要么整个操作完成，要么没开始，没有中间状态。

我个人习惯，能用 fetch_add 就别自己写 CAS 循环。后者虽然灵活，但性能开销更大，而且容易写出 bug。

12.2 memory_order 详解

很多初学者看到 memory_order 就头大。其实没那么玄乎，它控制的是指令重排的边界。

现代 CPU 和编译器为了性能，会打乱指令的执行顺序。单线程下没问题，多线程下就可能出乱子。比如：

std::atomic<bool> ready(false);
int data = 0;

// 线程1
data = 42;
ready.store(true, std::memory_order_release);

// 线程2
while (!ready.load(std::memory_order_acquire));
assert(data == 42);  // 这个断言能通过吗？

如果用了 release 和 acquire，就能保证线程2看到线程1写入的 data。这就是内存序的威力。

六种内存序速查表

内存序	含义	性能开销
memory_order_relaxed	只保证原子性，不保证顺序	几乎零开销
memory_order_consume	依赖关系排序（基本不用）	低
memory_order_acquire	后续读操作不能重排到前面	中等
memory_order_release	前面写操作不能重排到后面	中等
memory_order_acq_rel	acquire + release	较高
memory_order_seq_cst	全局顺序一致（默认）	最高

什么时候用 relaxed？

举个例子，你只是想统计某个函数被调用了多少次，不需要用它做同步：

std::atomic<int> call_count(0);

void my_function() {
    call_count.fetch_add(1, std::memory_order_relaxed);
    // 真正的业务逻辑...
}

这种情况下，relaxed 完全够用。你想想看，计数器多一次少一次无所谓，只要最终结果大致正确就行。但如果你用它来做「是否初始化」的判断，那就必须用 acquire/release 了。

12.3 避免 ABA 问题

ABA 问题，说白了就是：你 CAS 的时候，值从 A 变成 B 又变回 A，你误以为没人动过。

举个例子。你有个无锁栈，栈顶指针是 top。线程1读到 top = A，准备 CAS 把 top 改成 A->next。这时候线程2把 A 弹出了，又 push 了一个新节点 B，然后又把 B 弹出了，最后把 A 重新 push 回去。线程1的 CAS 发现 top 还是 A，就成功了——但此时 A->next 可能已经被释放了，或者指向了无效内存。

嗯，这就是 ABA 问题的典型场景。我在做无锁队列优化时，就踩过这个坑。

解决方案：标记指针

最常用的办法是给指针加一个版本号：

struct Node {
    int value;
    Node* next;
};

std::atomic<uintptr_t> top;  // 指针 + 版本号打包

void push(Node* new_node) {
    uintptr_t old_top = top.load();
    while (true) {
        new_node->next = reinterpret_cast<Node*>(old_top & ~0xFFFF);
        uintptr_t new_top = (old_top + 1) & 0xFFFF;  // 版本号加一
        new_top |= reinterpret_cast<uintptr_t>(new_node);
        if (top.compare_exchange_weak(old_top, new_top)) {
            break;
        }
    }
}

这里我们把指针的低 16 位用作版本号。每次 CAS 成功，版本号就加一。即使指针值相同，版本号不同，CAS 也会失败。

12.4 知识体系总览

下面这张图帮你理清本章的核心脉络：

12.5 避坑指南

最后分享几个我踩过的坑：

• 不要用 volatile 做原子操作。volatile 只防止编译器优化，不保证 CPU 级别的原子性。我在早期项目里犯过这个错，结果线上偶发数据错乱。
• CAS 循环记得加 pause 指令。在 x86 上，自旋等待时加 _mm_pause() 能减少功耗，提高性能。
• compare_exchange_weak 和 strong 的区别。weak 可能伪失败（spurious failure），但性能更好。循环 CAS 用 weak，单次 CAS 用 strong。

---