C++17 多线程系列（三）：原子操作与内存模型——从 volatile 到 memory_order
·
核心目标：理解
std::atomic的真正含义，掌握六种内存序，能用无锁数据结构替代简单场景下的互斥锁，建立对 CPU 内存模型的基本认知。前置知识：Part 1 的线程创建，Part 2 的 mutex 和 Data Race 概念。
3.1 为什么需要原子操作
3.1.1 volatile 不是原子操作
这是 C++ 中最常见的误解之一：
volatile int counter = 0; // ❌ 完全不能解决多线程问题!
// volatile 的含义:
// "这个变量可能被硬件/信号处理程序修改，编译器不要优化掉读写"
// 它跟多线程安全完全没有关系!
// 正确的多线程保护:
std::atomic<int> counter{0}; // ✅ 原子操作
// 或
int counter = 0;
std::mutex mtx; // ✅ 互斥锁保护
3.1.2 i++ 为什么不是原子的
int shared = 0;
void thread_func() {
shared++; // 看似一条语句，实际是三步
}
汇编视角 (x86-64):
mov eax, [shared] ; ① 加载: 从内存读入寄存器
inc eax ; ② 递增: 在寄存器中计算
mov [shared], eax ; ③ 存储: 写回内存
两个线程交错执行:
线程A执行①② 线程B执行①②③ 线程A执行③
→ 线程B的写操作被线程A的旧值覆盖 → 两次递增只增加一次
一句话：
volatile≠ 原子;i++≠ 原子。线程间共享变量必须使用atomic或mutex。
3.2 std::atomic 基础
3.2.1 支持的类型
| 类型 | 说明 | 常用操作 |
|---|---|---|
atomic<bool> |
布尔标志 | load / store / exchange |
atomic<int> / atomic<unsigned> 等 |
整数类型 | fetch_add / fetch_sub / CAS |
atomic<T*> |
指针 | fetch_add (指针算术) |
atomic<user_type> |
自定义类型 | 需满足 is_lock_free() |
atomic_flag |
最简原子布尔 | test_and_set / clear (保证 lock-free) |
3.2.2 基本操作
#include <atomic>
#include <cassert>
std::atomic<int> ai{0};
// ── 读写 ──
ai.store(42); // 写入
int v = ai.load(); // 读取
assert(v == 42);
// ── 交换 ──
int old = ai.exchange(100); // 设置新值，返回旧值
assert(old == 42);
// ── fetch_add / fetch_sub ──
std::atomic<int> counter{0};
int prev = counter.fetch_add(1); // 返回旧值，counter 变为 1
// 等价于 "return counter++;"
// 也可以 counter++ (operator++ 重载)
3.2.3 compare_exchange——CAS 循环
CAS（Compare-And-Swap）是无锁编程的核心原语：
std::atomic<int> x{10};
int expected = 10;
// compare_exchange_strong(expected, desired):
// 如果 x == expected → x = desired, 返回 true
// 否则 → expected = x, 返回 false
bool success = x.compare_exchange_strong(expected, 20);
// 结果: x = 20, expected = 10, success = true
// ── weak vs strong ──
// strong: 只在 x != expected 时失败
// weak: 可能"伪失败" (spurious failure)
// → weak 更适合 CAS 循环（伪失败后重试即可, 某些 CPU 上更快）
CAS 经典模式：
// 无锁递增
std::atomic<int> count{0};
void increment() {
int expected = count.load();
while (!count.compare_exchange_weak(expected, expected + 1)) {
// expected 已被更新为当前值, 重试即可
}
}
// 等价于 count.fetch_add(1), 但展示了 CAS 循环模式
3.2.4 is_lock_free
std::atomic<int> ai;
std::cout << std::boolalpha << ai.is_lock_free() << "\n";
// x86-64: true (int 原子操作硬件支持)
// 某些平台: false (需要 mutex 模拟)
// 编译期也可检查
if constexpr (std::atomic<int>::is_always_lock_free) {
// 编译期确定是 lock-free
}
3.3 整数原子操作
std::atomic<int> val{10};
// fetch_add / fetch_sub / fetch_and / fetch_or / fetch_xor
int old_add = val.fetch_add(5); // val=15, old_add=10
int old_sub = val.fetch_sub(3); // val=12, old_sub=15
int old_and = val.fetch_and(0xF); // val=12, old_and=12
int old_or = val.fetch_or(0xFF); // val=255, old_or=12
// 运算符重载 (等价于 memory_order_seq_cst 的 fetch_xxx)
val++; // fetch_add(1)
val--; // fetch_sub(1)
val += 5; // fetch_add(5)
无锁引用计数示例
class RefCounted {
std::atomic<int> ref_count_{1};
public:
void add_ref() { ref_count_.fetch_add(1, std::memory_order_relaxed); }
void release() {
if (ref_count_.fetch_sub(1, std::memory_order_acq_rel) == 1) {
delete this; // 最后一个引用, 安全删除
}
}
private:
~RefCounted() = default;
};
3.4 std::atomic_flag——自旋锁
atomic_flag 是最简单的原子类型，保证 lock-free，是自旋锁的标准材料：
class SpinLock {
std::atomic_flag flag_ = ATOMIC_FLAG_INIT;
public:
void lock() {
while (flag_.test_and_set(std::memory_order_acquire)) {
// 自旋: 反复尝试, 直到获得锁
// 在 x86 上可加 _mm_pause() 降低功耗
}
}
void unlock() {
flag_.clear(std::memory_order_release);
}
};
// 使用
SpinLock spin;
{
std::lock_guard<SpinLock> lock(spin); // 可以与 lock_guard 搭配!
// ... 临界区 ...
}
自旋锁 vs 互斥锁：
特性 SpinLock std::mutex 临界区极短 (< 100 周期) ✅ 高效 ❌ 系统调用开销大 临界区较长 ❌ 浪费 CPU ✅ 自动休眠 实时系统 ✅ 无休眠 ❌ 不可在中断中使用
3.5 内存序 memory_order 详解
3.5.1 什么是内存重排
编译器和 CPU 都可能对指令进行重排以提升性能——这对单线程透明，但多线程下可能产生反直觉的结果：
// Thread 1 // Thread 2
x.store(1, relaxed); if (y.load(relaxed) == 1) {
y.store(1, relaxed); assert(x.load(relaxed) == 1);
}
// 这个 assert 可能失败! 因为在 Thread 1 中 x 和 y 的写入可能被重排
3.5.2 六种内存序速查
| memory_order | 保证 | 开销 | 典型场景 |
|---|---|---|---|
relaxed |
仅原子性 | ⭐ | 计数器递增 |
acquire |
读后不重排 | ⭐⭐ | 读锁、读共享数据 |
release |
写前不重排 | ⭐⭐ | 写锁、发布数据 |
acq_rel |
读写均有序 | ⭐⭐⭐ | CAS 循环、fetch_sub |
seq_cst |
全局顺序一致 | ⭐⭐⭐⭐ | 默认（最安全） |
consume |
依赖有序 | ⭐ | 几乎不用 |
3.5.3 acquire-release 实战
// ── 用 atomic 实现一个轻量级 flag 通信 ──
std::atomic<bool> ready{false};
int data = 0; // 普通变量, 由 memory_order 保护
// Thread 1: Producer
void producer() {
data = 42; // ① 准备数据
ready.store(true, std::memory_order_release); // ② 发布: 保证 ① 对消费者可见
}
// Thread 2: Consumer
void consumer() {
while (!ready.load(std::memory_order_acquire)); // ③ 获取: 保证看到 release 前所有写入
assert(data == 42); // ④ 安全: 保证 data=42
}
3.5.4 如果不配对的后果
// ❌ 错误: 两边都用 relaxed
void bad_producer() {
data = 42;
ready.store(true, std::memory_order_relaxed); // 不保证 data 写入顺序!
}
void bad_consumer() {
while (!ready.load(std::memory_order_relaxed));
// data 可能是 0! (编译器和 CPU 重排了 data 和 ready 的顺序)
std::cout << data; // 未定义!
}
3.6 无锁编程入门
3.6.1 无锁栈（简化版）
#include <atomic>
#include <memory>
template <typename T>
class LockFreeStack {
struct Node {
T data;
Node* next;
Node(const T& d) : data(d), next(nullptr) {}
};
std::atomic<Node*> head_{nullptr};
public:
void push(const T& value) {
Node* new_node = new Node(value);
new_node->next = head_.load(std::memory_order_relaxed);
while (!head_.compare_exchange_weak(
new_node->next, // expected: 期望 head 没变
new_node, // desired: 新的 head
std::memory_order_release,
std::memory_order_relaxed)) {
// CAS 失败 → new_node->next 已被更新为最新 head → 重试
}
}
std::shared_ptr<T> pop() {
Node* old_head = head_.load(std::memory_order_relaxed);
while (old_head &&
!head_.compare_exchange_weak(
old_head, old_head->next,
std::memory_order_acquire,
std::memory_order_relaxed)) {
// 重试
}
if (old_head == nullptr) return nullptr;
auto result = std::make_shared<T>(old_head->data);
delete old_head; // 简化: 实际生产代码需处理 ABA 问题
return result;
}
};
3.6.2 ABA 问题简介
ABA 问题:
① Thread A 读到 head = 0x100 (指向 Node1)
② Thread B pop 了 Node1, 又 push 了 Node2, Node2 刚好分配到 0x100
③ Thread A 的 CAS 检查 head 仍是 0x100 → 通过!
④ 但 Node1 已经不是原来的 Node1 了 → 数据损坏
解决方案: tagged pointer (指针 + 版本号)
std::atomic<std::pair<Node*, uintptr_t>> head_;
每次修改 head 时递增版本号
3.7 原子操作 vs 互斥锁——性能对比
// 基准测试: 10 个线程各递增 100 万次
#include <benchmark/benchmark.h>
std::atomic<int64_t> atomic_counter{0};
int64_t mutex_counter = 0;
std::mutex mtx;
static void BM_Atomic(benchmark::State& state) {
for (auto _ : state) {
atomic_counter.fetch_add(1, std::memory_order_relaxed);
}
}
static void BM_Mutex(benchmark::State& state) {
for (auto _ : state) {
std::lock_guard lock(mtx);
++mutex_counter;
}
}
典型结果（10 线程并发，x86-64）：
| 方式 | 吞吐量 (ops/s) | 相对性能 |
|---|---|---|
relaxed atomic |
~200M | 最快 |
seq_cst atomic |
~80M | 2.5x 慢于 relaxed |
mutex (无竞争) |
~50M | 4x 慢于 relaxed |
mutex (有竞争) |
~5M | 40x 慢! |
结论：原子操作在简单递增场景下远超互斥锁。但如果临界区超过几条指令，mutex 的成本被临界区本身分摊，差异变小。
3.8 原子操作使用决策树
3.9 常见陷阱
| 陷阱 | 错误代码 | 正确做法 |
|---|---|---|
| volatile 当原子 | volatile int x; x++; |
std::atomic<int> x; |
| relaxed 同步 | 两边都 relaxed 做同步 |
acquire-release 配对 |
| 忘记配对 | 一边 release 一边 relaxed | release/acquire 必须成对 |
| CAS 不用循环 | 单次 compare_exchange 当原子 |
用 while 循环 + weak |
| 过保守用 seq_cst | 计数器也用 seq_cst |
降低到 relaxed |
3.10 小结
| 知识点 | 掌握程度 | 核心要点 |
|---|---|---|
| volatile ≠ atomic | 熟练 | volatile 用于硬件, atomic 用于多线程 |
| atomic 基本操作 | 掌握 | load/store/exchange/fetch_add/CAS |
| CAS 循环模式 | 掌握 | while + compare_exchange_weak |
| memory_order 六种 | 理解 | relaxed < acquire/release < seq_cst |
| acquire-release 配对 | 掌握 | release 发布, acquire 获取, 必须配对 |
| atomic_flag 自旋锁 | 理解 | 临界区极短时高效 |
| 无锁栈 | 了解 | CAS + ABA 问题 |
| 性能对比 | 掌握 | 简单操作 atomic 完胜, 复杂操作 mutex |
下期预告
[Part 4：异步编程] 将深入 C++ 异步体系：
std::future<T>—— 获取异步结果std::promise<T>—— 单向通信通道std::async—— 一行启动异步任务std::packaged_task—— 包装可调用对象std::shared_future—— 多消费者共享结果
推荐工具
-fsanitize=thread—— 即使 atomic 用错了也能检测perf stat -e cycles,instructions—— 对比 atomic 不同 memory_order 的 IPC- Compiler Explorer (godbolt.org) —— 直观查看不同 memory_order 的汇编差异
更多推荐
10
0- 0
已为社区贡献13条内容
所有评论(0)