如何理解 C++ 中的 atomic？

C++ 中的 atomic 深入理解

什么是 atomic？

std::atomic 是 C++11 引入的模板类，用于实现原子操作。原子操作是不可分割的操作，在多线程环境下不会被其他线程打断，保证了操作的原子性、可见性和有序性。

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

// ❌ 非原子操作：数据竞争
int counter = 0;
void unsafeIncrement() {
counter++; // 不是原子操作，可能导致数据竞争
}

// ✓ 原子操作：线程安全
std::atomic<int> atomicCounter(0);
void safeIncrement() {
atomicCounter++; // 原子操作，线程安全
}

为什么需要 atomic？

1. 解决数据竞争

#include <atomic>
#include <thread>
#include <vector>

// 数据竞争示例
void dataRaceExample() {
int value = 0;

// 多个线程同时修改 value
std::thread t1([&value]() {
for (int i = 0; i < 100000; ++i) {
value++; // ❌ 数据竞争
}
});

std::thread t2([&value]() {
for (int i = 0; i < 100000; ++i) {
value++; // ❌ 数据竞争
}
});

t1.join();
t2.join();

// 结果可能不是 200000，而是小于 200000 的某个值
std::cout << "Non-atomic result: " << value << "\\n";
}

// 使用 atomic 解决
void atomicExample() {
std::atomic<int> value(0);

std::thread t1([&value]() {
for (int i = 0; i < 100000; ++i) {
value++; // ✓ 原子操作
}
});

std::thread t2([&value]() {
for (int i = 0; i < 100000; ++i) {
value++; // ✓ 原子操作
}
});

t1.join();
t2.join();

// 结果一定是 200000
std::cout << "Atomic result: " << value << "\\n";
}

2. 内存序保证

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

std::atomic<bool> ready{false};
int data = 0;

void producer() {
data = 42; // 写入数据
ready = true; // 设置标志
}

void consumer() {
while (!ready) { // 等待标志
// 自旋等待
}
std::cout << "Data: " << data << "\\n"; // 读取数据
}

void memoryOrderExample() {
std::thread t1(producer);
std::thread t2(consumer);

t1.join();
t2.join();
}

atomic 的核心特性

1. 原子性（Atomicity）

操作要么完全执行，要么完全不执行，不会被中断。

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

// 以下操作都是原子的
value.store(10); // 原子存储
int x = value.load(); // 原子加载
value++; // 原子自增
value.fetch_add(5); // 原子加法
value.exchange(20); // 原子交换

2. 可见性（Visibility）

一个线程的修改对其他线程立即可见。

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

std::atomic<int> flag{0};

void thread1() {
flag.store(1, std::memory_order_release); // 释放语义
}

void thread2() {
while (flag.load(std::memory_order_acquire) != 1) {
// 获取语义，保证能看到 thread1 的所有写入
}
std::cout << "Flag is set\\n";
}

3. 有序性（Ordering）

保证操作的执行顺序，防止编译器和 CPU 重排序。

#include <atomic>
#include <thread>

std::atomic<int> x{0};
std::atomic<int> y{0};

void writeX() {
x.store(1, std::memory_order_release);
}

void writeY() {
y.store(1, std::memory_order_release);
}

void readXThenY() {
while (x.load(std::memory_order_acquire) != 1) {}
if (y.load(std::memory_order_acquire) == 0) {
std::cout << "Y is still 0\\n";
}
}

void readYThenX() {
while (y.load(std::memory_order_acquire) != 1) {}
if (x.load(std::memory_order_acquire) == 0) {
std::cout << "X is still 0\\n";
}
}

atomic 的基本操作

1. 基础操作

#include <atomic>
#include <iostream>

void basicOperations() {
std::atomic<int> value(0);

// store – 原子存储
value.store(10);
std::cout << "After store(10): " << value << "\\n";

// load – 原子加载
int x = value.load();
std::cout << "Loaded value: " << x << "\\n";

// exchange – 原子交换
int old = value.exchange(20);
std::cout << "Exchanged " << old << " with 20\\n";

// compare_exchange_weak – 弱 CAS
int expected = 20;
bool success = value.compare_exchange_weak(expected, 30);
std::cout << "CAS weak: " << (success ? "success" : "failed") << "\\n";

// compare_exchange_strong – 强 CAS
expected = 30;
success = value.compare_exchange_strong(expected, 40);
std::cout << "CAS strong: " << (success ? "success" : "failed") << "\\n";
}

2. 算术操作

void arithmeticOperations() {
std::atomic<int> value(10);

// fetch_add – 原子加法
int old = value.fetch_add(5);
std::cout << "fetch_add(5): old=" << old << ", new=" << value << "\\n";

// fetch_sub – 原子减法
old = value.fetch_sub(3);
std::cout << "fetch_sub(3): old=" << old << ", new=" << value << "\\n";

// fetch_and – 原子按位与
old = value.fetch_and(0x0F);
std::cout << "fetch_and(0x0F): old=" << old << ", new=" << value << "\\n";

// fetch_or – 原子按位或
old = value.fetch_or(0xF0);
std::cout << "fetch_or(0xF0): old=" << old << ", new=" << value << "\\n";

// fetch_xor – 原子按位异或
old = value.fetch_xor(0xFF);
std::cout << "fetch_xor(0xFF): old=" << old << ", new=" << value << "\\n";

// operator++ 和 operator–
++value; // 前置自增
value++; // 后置自增
—value; // 前置自减
value—; // 后置自减

// 复合赋值
value += 5;
value -= 3;
value &= 0x0F;
value |= 0xF0;
value ^= 0xFF;
}

3. 指针操作

void pointerOperations() {
int data[10] = {0, 1, 2, 3, 4, 5, 6, 7, 8, 9};
std::atomic<int*> ptr(data);

// 原子指针算术
int* old = ptr.fetch_add(2);
std::cout << "fetch_add(2): old=" << (old – data)
<< ", new=" << (ptr.load() – data) << "\\n";

old = ptr.fetch_sub(1);
std::cout << "fetch_sub(1): old=" << (old – data)
<< ", new=" << (ptr.load() – data) << "\\n";

// 前置/后置自增自减
++ptr; // ptr += 1
ptr++; // 返回旧值，然后 +1
—ptr; // ptr -= 1
ptr—; // 返回旧值，然后 -1
}

内存序（Memory Order）

内存序是 atomic 最复杂但也最重要的概念。

内存序类型

enum class memory_order {
relaxed, // 最宽松，只保证原子性
consume, // 消费语义（很少使用）
acquire, // 获取语义
release, // 释放语义
acq_rel, // 获取+释放
seq_cst // 顺序一致性（默认）
};

1. memory_order_relaxed

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

std::atomic<int> x{0};
std::atomic<int> y{0};

void relaxedExample() {
// relaxed 只保证原子性，不保证顺序
std::thread t1([]() {
x.store(1, std::memory_order_relaxed);
y.store(1, std::memory_order_relaxed);
});

std::thread t2([]() {
int r1 = y.load(std::memory_order_relaxed);
int r2 = x.load(std::memory_order_relaxed);
std::cout << "r1=" << r1 << ", r2=" << r2 << "\\n";
// 可能看到 r1=1, r2=0（重排序）
});

t1.join();
t2.join();
}

2. memory_order_acquire / memory_order_release

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

std::atomic<int> data{0};
std::atomic<bool> ready{false};

void acquireReleaseExample() {
// 生产者
std::thread producer([]() {
data.store(42, std::memory_order_relaxed); // 写入数据
ready.store(true, std::memory_order_release); // 释放语义
});

// 消费者
std::thread consumer([]() {
while (!ready.load(std::memory_order_acquire)) { // 获取语义
// 等待
}
// 保证能看到 data=42
std::cout << "Data: " << data.load(std::memory_order_relaxed) << "\\n";
});

producer.join();
consumer.join();
}

3. memory_order_seq_cst

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

std::atomic<int> x{0};
std::atomic<int> y{0};
std::atomic<int> z{0};

void seqCstExample() {
// 顺序一致性：所有线程看到相同的操作顺序
std::thread t1([]() {
x.store(1, std::memory_order_seq_cst);
});

std::thread t2([]() {
y.store(1, std::memory_order_seq_cst);
});

std::thread t3([]() {
int r1 = x.load(std::memory_order_seq_cst);
int r2 = y.load(std::memory_order_seq_cst);
std::cout << "Thread 3: r1=" << r1 << ", r2=" << r2 << "\\n";
});

std::thread t4([]() {
int r3 = y.load(std::memory_order_seq_cst);
int r4 = x.load(std::memory_order_seq_cst);
std::cout << "Thread 4: r3=" << r3 << ", r4=" << r4 << "\\n";
});

t1.join();
t2.join();
t3.join();
t4.join();
}

内存序选择指南

// 1. 简单计数器 – relaxed
std::atomic<int> counter{0};
counter.fetch_add(1, std::memory_order_relaxed);

// 2. 生产者-消费者模式 – acquire/release
std::atomic<bool> ready{false};
std::atomic<int> data{0};

// 生产者
data.store(value, std::memory_order_relaxed);
ready.store(true, std::memory_order_release);

// 消费者
while (!ready.load(std::memory_order_acquire)) {}
int value = data.load(std::memory_order_relaxed);

// 3. 需要全局顺序 – seq_cst（默认）
std::atomic<int> flag{0};
flag.store(1); // 默认是 seq_cst

CAS（Compare-And-Swap）操作

1. 基本用法

#include <atomic>
#include <iostream>

void casExample() {
std::atomic<int> value(10);

// compare_exchange_weak
int expected = 10;
if (value.compare_exchange_weak(expected, 20)) {
std::cout << "CAS succeeded, value is now " << value << "\\n";
} else {
std::cout << "CAS failed, expected is now " << expected << "\\n";
}

// compare_exchange_strong
expected = 20;
if (value.compare_exchange_strong(expected, 30)) {
std::cout << "CAS succeeded, value is now " << value << "\\n";
} else {
std::cout << "CAS failed, expected is now " << expected << "\\n";
}
}

2. weak vs strong

void weakVsStrong() {
std::atomic<int> value(10);

// weak: 可能失败（虚假失败），但性能更好
int expected = 10;
while (!value.compare_exchange_weak(expected, 20)) {
// 循环重试
}

// strong: 保证不会虚假失败，但性能稍差
expected = 20;
if (value.compare_exchange_strong(expected, 30)) {
std::cout << "Success\\n";
}
}

3. 无锁栈实现

#include <atomic>
#include <iostream>

template<typename T>
class LockFreeStack {
private:
struct Node {
T data;
Node* next;
Node(const T& value) : data(value), next(nullptr) {}
};

std::atomic<Node*> head;

public:
LockFreeStack() : head(nullptr) {}

void push(const T& value) {
Node* newNode = new Node(value);
newNode->next = head.load(std::memory_order_relaxed);

// CAS 操作
while (!head.compare_exchange_weak(
newNode->next,
newNode,
std::memory_order_release,
std::memory_order_relaxed
)) {
// 循环重试
}
}

bool pop(T& result) {
Node* oldHead = head.load(std::memory_order_acquire);

while (oldHead && !head.compare_exchange_weak(
oldHead,
oldHead->next,
std::memory_order_acquire,
std::memory_order_relaxed
)) {
// 循环重试
}

if (oldHead) {
result = oldHead->data;
delete oldHead;
return true;
}
return false;
}
};

void lockFreeStackExample() {
LockFreeStack<int> stack;

stack.push(1);
stack.push(2);
stack.push(3);

int value;
while (stack.pop(value)) {
std::cout << "Popped: " << value << "\\n";
}
}

atomic 与锁的对比

#include <atomic>
#include <mutex>
#include <thread>
#include <iostream>
#include <chrono>

// 使用 mutex
class CounterWithMutex {
private:
std::mutex mtx;
int value = 0;
public:
void increment() {
std::lock_guard<std::mutex> lock(mtx);
++value;
}

int get() const {
std::lock_guard<std::mutex> lock(mtx);
return value;
}
};

// 使用 atomic
class CounterWithAtomic {
private:
std::atomic<int> value{0};
public:
void increment() {
++value;
}

int get() const {
return value;
}
};

void performanceComparison() {
const int iterations = 1000000;

// 测试 mutex
CounterWithMutex counterMutex;
auto start1 = std::chrono::high_resolution_clock::now();

std::thread t1([&counterMutex, iterations]() {
for (int i = 0; i < iterations; ++i) counterMutex.increment();
});
std::thread t2([&counterMutex, iterations]() {
for (int i = 0; i < iterations; ++i) counterMutex.increment();
});

t1.join();
t2.join();

auto end1 = std::chrono::high_resolution_clock::now();
auto time1 = std::chrono::duration_cast<std::chrono::microseconds>(end1 – start1);

// 测试 atomic
CounterWithAtomic counterAtomic;
auto start2 = std::chrono::high_resolution_clock::now();

std::thread t3([&counterAtomic, iterations]() {
for (int i = 0; i < iterations; ++i) counterAtomic.increment();
});
std::thread t4([&counterAtomic, iterations]() {
for (int i = 0; i < iterations; ++i) counterAtomic.increment();
});

t3.join();
t4.join();

auto end2 = std::chrono::high_resolution_clock::now();
auto time2 = std::chrono::duration_cast<std::chrono::microseconds>(end2 – start2);

std::cout << "Mutex time: " << time1.count() << " μs\\n";
std::cout << "Atomic time: " << time2.count() << " μs\\n";
std::cout << "Speedup: " << (double)time1.count() / time2.count() << "x\\n";
}

atomic 的限制

1. 不是所有类型都支持

// ✓ 支持的类型
std::atomic<int> a1;
std::atomic<bool> a2;
std::atomic<float> a3;
std::atomic<int*> a4;

// ✗ 不支持复杂类型
// std::atomic<std::vector<int>> a5; // 编译错误
// std::atomic<std::string> a6; // 编译错误

// 但可以使用 is_lock_free 检查
std::atomic<int> x;
if (x.is_lock_free()) {
std::cout << "int is lock-free\\n";
} else {
std::cout << "int uses locks\\n";
}

2. 不能保证整个代码块的原子性

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

// ❌ 这不是原子的
if (value > 0) {
// 在这里，其他线程可能修改 value
value—; // 可能变成负数
}

// ✓ 正确做法：使用 CAS
int expected = value.load();
while (expected > 0 && !value.compare_exchange_weak(expected, expected – 1)) {
// 循环重试
}

3. ABA 问题

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

std::atomic<int*> ptr(nullptr);

void abaProblem() {
int* x = new int(1);
int* y = new int(2);

ptr.store(x);

std::thread t1([&]() {
int* p = ptr.load();
int* next = p ? new int(*p + 1) : nullptr;
// ABA 问题：p 可能被其他线程修改后又改回来
if (ptr.compare_exchange_weak(p, next)) {
std::cout << "CAS succeeded\\n";
} else {
std::cout << "CAS failed\\n";
delete next;
}
});

std::thread t2([&]() {
int* p = ptr.load();
delete p;
ptr.store(y);
// 又改回 x
ptr.store(new int(1));
});

t1.join();
t2.join();
}

实际应用场景

1. 引用计数

#include <atomic>
#include <iostream>

class SharedObject {
private:
std::atomic<int> refCount{1};

public:
void addRef() {
refCount.fetch_add(1, std::memory_order_relaxed);
}

void release() {
if (refCount.fetch_sub(1, std::memory_order_acq_rel) == 1) {
delete this;
}
}

int getRefCount() const {
return refCount.load(std::memory_order_relaxed);
}

virtual ~SharedObject() {
std::cout << "Object destroyed\\n";
}
};

void referenceCountingExample() {
SharedObject* obj = new SharedObject();
std::cout << "Initial ref count: " << obj->getRefCount() << "\\n";

obj->addRef();
std::cout << "After addRef: " << obj->getRefCount() << "\\n";

obj->release();
std::cout << "After release: " << obj->getRefCount() << "\\n";

obj->release(); // 对象被销毁
}

2. 单例模式

#include <atomic>
#include <iostream>

class Singleton {
private:
static std::atomic<Singleton*> instance;
static std::mutex mtx;

Singleton() {
std::cout << "Singleton created\\n";
}

public:
static Singleton* getInstance() {
Singleton* tmp = instance.load(std::memory_order_acquire);
if (tmp == nullptr) {
std::lock_guard<std::mutex> lock(mtx);
tmp = instance.load(std::memory_order_relaxed);
if (tmp == nullptr) {
tmp = new Singleton();
instance.store(tmp, std::memory_order_release);
}
}
return tmp;
}

void doSomething() {
std::cout << "Doing something\\n";
}
};

std::atomic<Singleton*> Singleton::instance{nullptr};
std::mutex Singleton::mtx;

void singletonExample() {
Singleton* s1 = Singleton::getInstance();
Singleton* s2 = Singleton::getInstance();

std::cout << "s1 == s2: " << (s1 == s2) << "\\n";
s1->doSomething();
}

3. 自旋锁

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

class SpinLock {
private:
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);
}
};

SpinLock spinLock;
int sharedData = 0;

void spinLockExample() {
auto worker = [](int id) {
for (int i = 0; i < 10000; ++i) {
spinLock.lock();
++sharedData;
spinLock.unlock();
}
};

std::thread t1(worker, 1);
std::thread t2(worker, 2);

t1.join();
t2.join();

std::cout << "Shared data: " << sharedData << "\\n";
}

总结

使用建议：

std::atomic 是 C++ 并发编程的基础工具，正确理解和使用它对于编写高效、安全的并发程序至关重要。