多线程里counter的结果为什么会不对因为在机器层面是读、改、写三步而这三步之间随时可能被另一个线程插进来。std::atomicT把这一整套变成不可分割的一步代价是你要理解它到底保证了什么、没保证什么。这篇从丢更新这个现象切入把is_lock_free真跑出来看讲清compare_exchange_weak为什么是无锁数据结构的基石最后用实测数据回答那个高频问题std::atomic和volatile到底是不是一回事。1. 引子counter丢在了哪里先看一段错误的代码。它没有编译错误、没有警告但结果是错的// 反例不要这么写数据竞争Core Guidelines 要求共享可变状态必须同步 #include thread #include vector int counter 0; // 普通 int不是原子类型 void work() { for (int k 0; k 100000; k) { counter; // ① 读 counter ② 加一 ③ 写回 —— 三步随时可被打断 } } // 真实程序里会这样跑起 4 个线程各调一次 work()然后 join。 // 期望得到 400000实测通常是一个小于 400000 的随机数、每次都不一样 // —— 结果不稳定所以这里不给输出块正确版本见第 6 节 void demo() { std::vectorstd::thread workers; for (int t 0; t 4; t) workers.emplace_back(work); for (auto t : workers) t.join(); }它错在counter不是一步操作非原子 counter4 线程同时跑—— 丢失更新lost update 时间 ──► 线程 A ── 读 counter100 ────────────────────────────── 写 counter101 ──► 线程 B ── 读 counter100 ── 写 counter101 ──► ▲ └── B 的写覆盖了 A 的写 两次 只让 counter 增加了 1 原子 counterstd::atomicint读-改-写是一个不可分割的 RMW 操作 线程 A ── [ 读 100 → 改 101 → 写 101 ] ─────────────────────────────► 线程 B ── [ 101 → 102 → 102 ] ──► 结果400000且每次运行都一样这段反例的最终值不稳定取决于调度所以这篇不给它的输出它的正确版本在第 5 节会打印确定的400000。要提醒的是这段代码不只是「结果可能变小」在 C 里对同一个非原子对象并发读写本身就是未定义行为undefined behavior编译器完全有理由把它优化成任何东西只是它通常看起来「只是少算了一点」反而更危险。官方文档std::atomic — cppreference · C Core Guidelines CP.8「不要用 volatile 做同步」2. 原子性保证了什么is_lock_free又是什么std::atomicT的保证可以浓缩成两条不可分割atomicity对同一个原子对象的所有操作在别的线程看来是「要么全发生、要么全没发生」绝不会出现撕裂读torn read读到一半新一半旧的中间状态。可见性与定序默认参数下memory_order_seq_cst还提供跨线程的同步与顺序保证。注意它只保护这一个对象本身。std::atomicT不会保护 T 里指针指向的内存也不会保护「两个原子变量之间的一致性」那是《memory_order 六种内存序详解》要解决的问题。第二件事是is_lock_free()它回答「这个原子类型在当前平台上是不是真的用无锁指令实现的」。std::atomicint在 x86-64 上就是一条lock xadd而某些平台/某些宽度上它会退化成内部锁内部加一把锁来模拟原子性性能大约等于 mutex。is_always_lock_free是编译期常量可以拿来做static_assert。真跑一遍看// atomic_lockfree.cpp — 编译: g -stdc17 -Wall -O2 -pthread atomic_lockfree.cpp -o alf #include atomic #include cstdio struct Pair16 { long long a; long long b; }; // 16 字节 struct Big { char data[64]; }; // 64 字节 int main() { std::atomicchar ac{0}; std::atomicshort as{0}; std::atomicint ai{0}; std::atomiclong long all{0}; std::atomicdouble ad{0.0}; std::atomicint* ap{nullptr}; std::printf(char %2zu 字节 is_lock_free%d\n, sizeof(ac), static_castint(ac.is_lock_free())); std::printf(short %2zu 字节 is_lock_free%d\n, sizeof(as), static_castint(as.is_lock_free())); std::printf(int %2zu 字节 is_lock_free%d\n, sizeof(ai), static_castint(ai.is_lock_free())); std::printf(long long %2zu 字节 is_lock_free%d\n, sizeof(all), static_castint(all.is_lock_free())); std::printf(double %2zu 字节 is_lock_free%d\n, sizeof(ad), static_castint(ad.is_lock_free())); std::printf(int* %2zu 字节 is_lock_free%d\n, sizeof(ap), static_castint(ap.is_lock_free())); std::printf(is_always_lock_free(int) %d\n, static_castint(std::atomicint::is_always_lock_free)); std::printf(is_always_lock_free(16B) %d\n, static_castint(std::atomicPair16::is_always_lock_free)); std::printf(is_always_lock_free(64B) %d\n, static_castint(std::atomicBig::is_always_lock_free)); }char 1 字节 is_lock_free1 short 2 字节 is_lock_free1 int 4 字节 is_lock_free1 long long 8 字节 is_lock_free1 double 8 字节 is_lock_free1 int* 8 字节 is_lock_free1 is_always_lock_free(int) 1 is_always_lock_free(16B) 0 is_always_lock_free(64B) 0结论8 字节及以下的标量类型在 x86-64 上全部无锁硬件直接支持lock cmpxchg/lock xadd而 16 字节 和 64 字节的结构体都不是。16 字节本可以靠cmpxchg16b做到但那需要编译器显式开启对应指令集64 字节则无论如何都要退化成锁。类型宽度典型实现is_lock_free()性能特征1 / 2 / 4 / 8 字节标量、指针单条lock前缀指令true纳秒级无内核参与16 字节结构体默认编译选项内部锁 /libatomicfalse实测有一次内部互斥量竞争64 字节及更大的结构体内部锁false与 mutex 接近通常不如直接用锁有个实测细节很能说明问题如果你在文章里把std::atomicBig::is_lock_free()真的调用出来并链接会得到undefined reference to __atomic_is_lock_free因为大类型的实现已经被挪进libatomic库需要加-latomic这本身就说明它不再是纯内联的无锁指令而是一个库函数调用。所以实践结论很干脆要无锁就把原子类型控制在 8 字节以内需要原子地更新一大坨数据就老老实实用锁 值语义。官方文档std::atomic::is_lock_free — cppreference · std::atomic::is_always_lock_free — cppreference3. 常用成员一览std::atomicT的接口不多但每个都有明确的使用场景成员作用适合的场景备注load()原子读读标志位、读计数器快照相当于operator T()store(v)原子写设标志位、发布新值相当于operator(v)exchange(v)原子交换返回旧值「取走并清空」、抢令牌一条指令完成fetch_add(v)/fetch_sub(v)原子加减返回旧值计数、引用计数、序号分配/--也是 RMWfetch_or/and/xor原子位运算位图、标志集合主要用于整数compare_exchange_weak/strongCAS无锁数据结构、原子更新复杂值无锁编程的基石is_lock_free()运行时查询性能调优、平台适配见上一节fetch_add返回的是旧值这一点常被忽略想做「我是第几个」的序号分配auto id next_id.fetch_add(1);拿到的id就是自己的序号不需要再加一。4.compare_exchange无锁编程的基石CAScompare-and-swap的语义是「如果当前值等于我期望的旧值就把它换成新值并返回 true否则把当前真实值写回我的期望变量并返回 false」。第二半句是它最精妙的地方失败时顺手帮你刷新了「最新值」所以循环里不需要重新load// CAS 的语义伪代码 bool compare_exchange(int expected, int desired) { if (value expected) { value desired; return true; } expected value; // ← 失败时把真实值刷进 expected return false; }正因为它失败会刷新CAS循环才写得那么紧凑。下面用它实现一个「原子地更新最大值」四个线程各报一批数最终最大值必然确定// atomic_cas.cpp — 编译: g -stdc17 -Wall -O2 -pthread atomic_cas.cpp -o acas #include atomic #include cstdio #include thread #include vector void update_max(std::atomicint target, int value) { int current target.load(); while (current value !target.compare_exchange_weak(current, value)) { // 失败时 current 已被刷新成真实值直接进入下一轮比较 } } int main() { constexpr int kThreads 4; constexpr int kRounds 5000; constexpr int kPeak 1000; std::atomicint best{0}; std::vectorstd::thread workers; for (int t 0; t kThreads; t) { workers.emplace_back([best, t] { for (int k 0; k kRounds; k) { update_max(best, (t * kRounds k) % kPeak 1); // 取值在 1..1000 之间循环 } }); } for (auto t : workers) t.join(); std::printf(并发更新后最大值 %d\n, best.load()); best.store(7); int expected 7; const bool ok best.compare_exchange_strong(expected, 9); std::printf(CAS(7 - 9) 成功 %dbest %d\n, static_castint(ok), best.load()); int stale 3; const bool failed best.compare_exchange_strong(stale, 5); std::printf(CAS(3 - 5) 成功 %dbest %dstale 被刷新为 %d\n, static_castint(failed), best.load(), stale); }并发更新后最大值 1000 CAS(7 - 9) 成功 1best 9 CAS(3 - 5) 成功 0best 9stale 被刷新为 9看第三行CAS 期望「当前是 3」实际是 9于是失败、返回 0并且把stale从 3 刷成了 9。这就是上面说的「失败顺手刷新」也是为什么update_max里不需要在循环里重新loadcurrent已经被刷成最新值了。weak和strong的区别只有一个weak允许「值确实相等但依然失败」伪失败spurious failurestrong不允许。所以在循环里用weak省一次指令失败了重试就行在「只试一次」的场合用strong。变体伪失败适用场合compare_exchange_weak可能即使值相等写在循环里性能略好compare_exchange_strong不会只尝试一次、或循环体很贵不宜重试5.std::atomic与volatile的真正区别这是 C 并发里传播最广的误解之一volatile不提供任何同步也不保证原子性。它的原意是「这块内存可能被编译器看不见的力量改变」所以禁止编译器把访问优化掉、禁止跨访问重排服务的是内存映射 I/O 和信号处理跟多线程毫无关系。下面用一次被强行排定顺序的「丢失更新」把它演出来两个线程都先读到旧值再一起写回volatile一点忙都帮不上。// volatile_lost_update.cpp — 编译: g -stdc17 -Wall -O2 -pthread volatile_lost_update.cpp -o vlu // 反例volatile 不能替代原子操作。这段代码本身也含数据竞争只为演示「丢失更新」。 #include atomic #include cstdio #include thread volatile int g_counter 0; // 反例volatile 不提供原子性Core Guidelines CP.8 std::atomicint g_read{0}; // 用它们来强制出一个确定的交错顺序 std::atomicbool g_can_write{false}; int main() { std::thread a([] { const int value g_counter; // ① 读 g_read.fetch_add(1); while (!g_can_write.load()) { } // ② 等另一个线程也读完 g_counter value 1; // ③ 写 }); std::thread b([] { const int value g_counter; g_read.fetch_add(1); while (!g_can_write.load()) { } g_counter value 1; }); while (g_read.load() ! 2) { } // 两个线程都读到旧值了 g_can_write.store(true); // 才允许它们写回 a.join(); b.join(); std::printf(两次 1 后 g_counter %d正确应为 2\n, g_counter); }两次 1 后 g_counter 1正确应为 2两次1只让计数器动了 1一次更新被完整地覆盖掉了。把volatile int换成std::atomicint同样被排定的交错下结果是 2因为g_counter value 1这种「先读后写」在原子类型上要写成fetch_add或者 CAS 循环两步之间无法被插入。维度volatilestd::atomicT原子性不保证读改写可被打断保证不可分割的 RMW跨线程同步 / 内存序完全不提供可指定memory_order默认seq_cst禁止编译器优化掉访问是是因为它有语义编译器本来就不能删是否会引起重排不保证对非 volatile 访问无约束通过内存序约束能不能当跨线程标志位 / 计数器不能能正当用途内存映射寄存器、信号处理、setjmp缓冲区计数器、标志位、无锁数据结构一句话记volatile管的是「编译器别乱优化我的访问」std::atomic管的是「多个线程别互相打架」。这两件事没有任何重叠。6.atomicbool做停止标志以及性能对比最常见的原子用法不是计数器而是一个停止标志stop flag一个线程干活另一个线程随时叫停。它必须是原子的否则干活线程可能永远读不到新值编译器可以把它提升到寄存器里。// atomic_stop.cpp — 编译: g -stdc17 -Wall -O2 -pthread atomic_stop.cpp -o astop #include atomic #include cstdio #include thread #include vector int main() { constexpr int kThreads 4; constexpr int kRounds 100000; std::atomicint counter{0}; std::vectorstd::thread workers; for (int t 0; t kThreads; t) { workers.emplace_back([counter] { for (int k 0; k kRounds; k) { counter; // 原子读-改-写 } }); } for (auto t : workers) t.join(); std::printf(std::atomicint 自增 %d 次 %d\n, kThreads * kRounds, counter.load()); std::printf(counter.is_lock_free() %d\n, static_castint(counter.is_lock_free())); std::atomicbool stop{false}; // 停止标志 std::atomiclong long ticks{0}; std::thread spinner([stop, ticks] { while (!stop.load(std::memory_order_relaxed)) { ticks.fetch_add(1, std::memory_order_relaxed); } }); std::thread stopper([stop] { stop.store(true, std::memory_order_relaxed); }); stopper.join(); spinner.join(); std::printf(停止标志生效 %dticks 0 %d\n, static_castint(stop.load()), static_castint(ticks.load() 0)); }std::atomicint 自增 400000 次 400000 counter.is_lock_free() 1 停止标志生效 1ticks 0 1ticks的具体值当然不确定取决于调度所以只打印ticks 0这个确定的事实确定性结果优先是并发示例的写作准则。原子自增 vs mutex 保护的int同样 400 万次自增一个用原子操作、一个用lock_guard保护普通int// atomic_vs_mutex.cpp — 编译: g -stdc17 -Wall -O2 -pthread atomic_vs_mutex.cpp -o avm #include atomic #include chrono #include cstdio #include mutex #include thread #include vector int main() { constexpr int kThreads 4; constexpr int kRounds 1000000; std::atomiclong long atomic_counter{0}; long long mutex_counter 0; std::mutex mutex; const auto t0 std::chrono::steady_clock::now(); { std::vectorstd::thread workers; for (int t 0; t kThreads; t) { workers.emplace_back([atomic_counter] { for (int k 0; k kRounds; k) { atomic_counter.fetch_add(1, std::memory_order_relaxed); } }); } for (auto t : workers) t.join(); } const auto t1 std::chrono::steady_clock::now(); { std::vectorstd::thread workers; for (int t 0; t kThreads; t) { workers.emplace_back([mutex_counter, mutex] { for (int k 0; k kRounds; k) { std::lock_guardstd::mutex lock(mutex); mutex_counter; } }); } for (auto t : workers) t.join(); } const auto t2 std::chrono::steady_clock::now(); const auto atomic_ms std::chrono::duration_caststd::chrono::milliseconds(t1 - t0).count(); const auto mutex_ms std::chrono::duration_caststd::chrono::milliseconds(t2 - t1).count(); std::printf(atomic 计数 %lld耗时 %lld ms\n, atomic_counter.load(), static_castlong long(atomic_ms)); std::printf(mutex 计数 %lld耗时 %lld ms\n, mutex_counter, static_castlong long(mutex_ms)); }atomic 计数 4000000耗时 ~~ ms mutex 计数 4000000耗时 ~~ ms在这台机器上实测大概是36 ms vs 121 ms原子版本快约 3.4 倍。耗时随机器和负载浮动所以这里用占位符但量级上的差距是稳定的。为什么差这么多不是「原子更省指令」那么简单对比项atomic.fetch_add(relaxed)lock_guardint单次操作一条lock xadd约 20 个周期锁总线/缓存行一条lock cmpxchg加锁 一条lock cmpxchg解锁 循环判断是否进入内核不进入不进入无竞争时有竞争时会 futex 系统调用竞争加剧时纯用户态自旋重试不会睡等待者会在内核里挂起唤醒要跨上下文切换内存屏障一条lock前缀自带的顺序约束每次加解锁都带全屏障语义能力只能操作这一个值能保护一整段临界区、多个变量关键差异在第 3 行mutex 无竞争时确实很快但一旦出现竞争失败者会被内核挂起唤醒它要走一次上下文切换那是微秒级的开销比一次原子自旋贵两到三个数量级。而fetch_add的失败只是一次缓存行失效后的重试全程待在用户态。但别急着把所有锁都换成原子这张表最后一行才是重点atomic只能保护「一个值」一旦你的不变量涉及两个以上的变量比如「队列非空」加「队列长度」就必须用锁。选型顺序是能用一个 8 字节以内的原子量表达 → 用原子否则 → 用锁并且优先考虑能不能改成「不共享」。7. 完整示例并发统计一批值把 sum / max / count 三种统计放在一起sum和count用fetch_add天然可加max没有「原子取最大」指令只能用 CAS 循环。四线程各处理一万个值输出全部确定// atomic_final.cpp — 编译: g -stdc17 -Wall -O2 -pthread atomic_final.cpp -o af #include atomic #include cstdio #include thread #include vector // 原子取最大值没有对应指令只能 CAS 循环 void update_max(std::atomicint target, int value) { int current target.load(std::memory_order_relaxed); while (current value !target.compare_exchange_weak(current, value, std::memory_order_relaxed, std::memory_order_relaxed)) { // 失败时 current 被刷新继续比较 } } int main() { constexpr int kThreads 4; constexpr int kRounds 10000; constexpr int kMod 1000; // 取值 1..1000 循环 std::atomiclong long sum{0}; std::atomicint max_seen{0}; std::atomicint processed{0}; std::vectorstd::thread workers; for (int t 0; t kThreads; t) { workers.emplace_back([, t] { for (int k 0; k kRounds; k) { const int value (t * kRounds k) % kMod 1; sum.fetch_add(value, std::memory_order_relaxed); // 可加直接 RMW update_max(max_seen, value); // 不可加CAS 循环 processed.fetch_add(1, std::memory_order_relaxed); } }); } for (auto t : workers) t.join(); std::printf(线程数 %d每线程处理 %d 个值\n, kThreads, kRounds); std::printf(处理总数 %d\n, processed.load()); std::printf(总和 %lld\n, sum.load()); std::printf(最大值 %d\n, max_seen.load()); std::printf(is_lock_free: long long%d, int%d\n, static_castint(std::atomiclong long::is_always_lock_free), static_castint(std::atomicint::is_always_lock_free)); }线程数 4每线程处理 10000 个值 处理总数 40000 总和 20020000 最大值 1000 is_lock_free: long long1, int1四个数字都确定40000 4 × 1000020020000 4 × 10 × (12…1000)每个线程都完整覆盖了 1..1000 十轮最大值 1000 必然出现。注意update_max里 CAS 的循环条件外层while判断current value只有「最大值确实更小」时才去 CAS这样绝大多数调用在第一次比较就退出了完全没有 CAS 的开销。这是compare_exchange的常见优化套路先用普通负载判断再 CAS。官方文档compare_exchange — cppreference · std::memory_order — cppreference《memory_order 六种内存序详解relaxed、acquire、release 到底怎么选》会专门展开8. 延伸阅读std::atomic — cppreference —— 模板总览要求T可平凡复制列出全部成员函数与特化std::atomic::compare_exchange — cppreference ——weak/strong的差别与「失败时刷新 expected」的正式措辞std::atomic::is_lock_free — cppreference —— 运行时查询配is_always_lock_free使用std::atomic_flag — cppreference —— 保证无锁的最小自旋锁原语test_and_set与 C20 的wait/notify_oneC Core Guidelines · 并发章节 CP.8 —— 明确写了「不要用 volatile 做同步」这篇第 5 节的结论就出自这里本知识库内的相关篇目《memory_order 六种内存序详解relaxed、acquire、release 到底怎么选》 —— 加了 std::atomic 结果还是错的因为原子性只保证「这一步不可分割」《mutex 与 lock_guard / unique_lock共享数据的正确加锁姿势》 —— 多个线程改同一个变量会怎样这篇先讲清数据竞争读-改-写被打断为什么让结果随机《shared_mutex 读写锁该不该用以及 volatile 的三大误解》 —— C17 的 std::shared_mutex 让多个读者并存、写者独占9. 一句话总结std::atomicT把「读-改-写」变成不可分割的一步并保证跨线程可见性 —— 它只保护这一个对象不保护它指向的东西、也不维护多个原子量之间的一致性8 字节以内的标量在 x86-64 上都是无锁的is_lock_free()实测为 1再大就退化成内部锁甚至需要libatomiccompare_exchange失败时会把真实值刷回expected这是无锁循环能写得那么短的原因而volatile和它毫无重叠 ——volatile只管「编译器别优化掉我的访问」不做同步、不保证原子性拿它当计数器必然丢更新。性能上原子自增比 mutex 保护的快数倍差距来自「竞争时 mutex 要把失败者挂进内核、唤醒要走上下文切换而原子操作全程待用户态自旋重试」。