当前位置：首页 > news >正文

原子操作：多线程编程

news 2025/9/18 12:21:37

🎯 什么是原子操作

原子操作（Atomic Operation）是指在执行过程中不会被其他线程打断的操作，它要么完全执行，要么完全不执行，不存在中间状态。在多线程编程中，原子操作是实现线程安全的重要工具。

为什么需要原子操作？

考虑一个简单的递增操作：

// ❌ 非线程安全的操作
int counter = 0;// 线程A和线程B同时执行
counter++;  // 这不是原子操作！

counter++ 实际上包含三个步骤：

读取 counter 的值到寄存器
计算寄存器值 + 1
写回结果到 counter

如果两个线程同时执行，可能出现：

线程A读取 counter = 0
线程B读取 counter = 0
线程A计算并写回 counter = 1
线程B计算并写回 counter = 1

结果：期望值是2，实际值是1！

🔧 C++中的原子操作

std::atomic 基础用法

#include <atomic>
#include <thread>
#include <iostream>
#include <vector>class ThreadSafeCounter {
private:std::atomic<int> count{0};  // ✅ 原子整数public:void increment() {count++;  // ✅ 原子递增操作}void decrement() {count--;  // ✅ 原子递减操作}int get() const {return count.load();  // ✅ 原子读取}void set(int value) {count.store(value);   // ✅ 原子写入}// 比较并交换bool compareAndSwap(int expected, int newValue) {return count.compare_exchange_weak(expected, newValue);}
};// 使用示例
void testAtomicCounter() {ThreadSafeCounter counter;std::vector<std::thread> threads;// 创建10个线程，每个线程递增1000次for (int i = 0; i < 10; ++i) {threads.emplace_back([&counter]() {for (int j = 0; j < 1000; ++j) {counter.increment();}});}// 等待所有线程完成for (auto& t : threads) {t.join();}std::cout << "Final count: " << counter.get() << std::endl;  // 输出：10000
}

🎯 常用的原子操作类型

1. 基本算术操作

std::atomic<int> atomicInt{10};// 原子递增/递减
int oldValue = atomicInt++;      // 后置递增，返回旧值
int newValue = ++atomicInt;      // 前置递增，返回新值
atomicInt += 5;                  // 原子加法
atomicInt -= 3;                  // 原子减法
atomicInt *= 2;                  // 原子乘法
atomicInt /= 2;                  // 原子除法// fetch_xxx 系列（返回操作前的值）
int oldVal = atomicInt.fetch_add(10);    // 加10，返回旧值
int oldVal2 = atomicInt.fetch_sub(5);    // 减5，返回旧值

2. 比较并交换 (CAS - Compare And Swap)

std::atomic<int> atomicValue{5};int expected = 5;
int newValue = 10;// weak版本：可能会因为假失败而返回false
if (atomicValue.compare_exchange_weak(expected, newValue)) {std::cout << "Successfully changed from 5 to 10" << std::endl;
}// strong版本：只有在真的不相等时才返回false
expected = 10;
int anotherValue = 20;
if (atomicValue.compare_exchange_strong(expected, anotherValue)) {std::cout << "Successfully changed from 10 to 20" << std::endl;
} else {std::cout << "Current value is: " << expected << std::endl;  // expected被更新为实际值
}

3. 内存序（Memory Ordering）

std::atomic<bool> ready{false};
std::atomic<int> data{0};// 生产者线程
void producer() {data.store(42, std::memory_order_relaxed);          // 设置数据ready.store(true, std::memory_order_release);       // 发布信号
}// 消费者线程
void consumer() {while (!ready.load(std::memory_order_acquire)) {    // 等待信号// 自旋等待}std::cout << "Data: " << data.load(std::memory_order_relaxed) << std::endl;  // 读取数据
}

📋 实际应用场景

场景1：线程安全的单例模式

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

场景2：无锁队列（简化版）

template<typename T>
class LockFreeQueue {
private:struct Node {std::atomic<T*> data{nullptr};std::atomic<Node*> next{nullptr};};std::atomic<Node*> head{new Node};std::atomic<Node*> tail{head.load()};public:void enqueue(T item) {Node* newNode = new Node;T* data = new T(std::move(item));newNode->data.store(data);Node* prevTail = tail.exchange(newNode);prevTail->next.store(newNode);}bool dequeue(T& result) {Node* head_node = head.load();Node* next = head_node->next.load();if (next == nullptr) {return false;  // 队列为空}T* data = next->data.load();if (data == nullptr) {return false;}result = *data;delete data;head.store(next);delete head_node;return true;}
};

场景3：线程池中的任务计数

class ThreadPool {
private:std::atomic<size_t> activeTasks{0};std::atomic<bool> shutdown{false};std::vector<std::thread> workers;public:void submitTask(std::function<void()> task) {if (shutdown.load()) return;activeTasks.fetch_add(1);  // 原子递增任务计数// 提交任务给工作线程...auto wrappedTask = [this, task]() {try {task();} catch (...) {// 处理异常}activeTasks.fetch_sub(1);  // 任务完成，原子递减};// 将 wrappedTask 放入队列...}void waitForAllTasks() {while (activeTasks.load() > 0) {std::this_thread::sleep_for(std::chrono::milliseconds(1));}}size_t getActiveTaskCount() const {return activeTasks.load();}
};

场景4：状态标志管理

class NetworkConnection {
private:enum State { DISCONNECTED, CONNECTING, CONNECTED, ERROR };std::atomic<State> connectionState{DISCONNECTED};std::atomic<bool> shouldStop{false};public:void connect() {State expected = DISCONNECTED;if (connectionState.compare_exchange_strong(expected, CONNECTING)) {// 执行连接逻辑...if (/* 连接成功 */) {connectionState.store(CONNECTED);} else {connectionState.store(ERROR);}}}void disconnect() {shouldStop.store(true);connectionState.store(DISCONNECTED);}bool isConnected() const {return connectionState.load() == CONNECTED;}bool shouldStopOperation() const {return shouldStop.load();}
};

🎯 Qt中的原子操作

Qt提供了自己的原子操作类：

#include <QAtomicInt>
#include <QAtomicPointer>class QtAtomicExample {
private:QAtomicInt counter{0};QAtomicPointer<QString> message{nullptr};public:void increment() {counter.fetchAndAddAcquire(1);}int getCount() const {return counter.loadAcquire();}void setMessage(QString* msg) {QString* old = message.fetchAndStoreRelease(msg);delete old;  // 安全删除旧消息}QString* getMessage() const {return message.loadAcquire();}
};

⚡ 性能考虑

原子操作 vs 互斥锁

#include <chrono>
#include <mutex>// 性能测试：原子操作 vs 互斥锁
class PerformanceTest {
private:std::atomic<long long> atomicCounter{0};long long mutexCounter{0};std::mutex mtx;public:void testAtomic(int iterations) {auto start = std::chrono::high_resolution_clock::now();for (int i = 0; i < iterations; ++i) {atomicCounter.fetch_add(1);}auto end = std::chrono::high_resolution_clock::now();auto duration = std::chrono::duration_cast<std::chrono::microseconds>(end - start);std::cout << "Atomic operations: " << duration.count() << " μs" << std::endl;}void testMutex(int iterations) {auto start = std::chrono::high_resolution_clock::now();for (int i = 0; i < iterations; ++i) {std::lock_guard<std::mutex> lock(mtx);++mutexCounter;}auto end = std::chrono::high_resolution_clock::now();auto duration = std::chrono::duration_cast<std::chrono::microseconds>(end - start);std::cout << "Mutex operations: " << duration.count() << " μs" << std::endl;}
};

📝 最佳实践

1. 选择合适的内存序

// 对于简单的计数器，relaxed足够
std::atomic<int> counter{0};
counter.fetch_add(1, std::memory_order_relaxed);// 对于同步操作，使用acquire-release
std::atomic<bool> ready{false};
ready.store(true, std::memory_order_release);  // 生产者
while (!ready.load(std::memory_order_acquire)) {} // 消费者

2. 避免ABA问题

// 使用版本号避免ABA问题
struct VersionedPointer {void* pointer;size_t version;
};std::atomic<VersionedPointer> versionedPtr;bool safeCompareExchange(void* expected, void* newPtr) {VersionedPointer current = versionedPtr.load();if (current.pointer == expected) {VersionedPointer newValue = {newPtr, current.version + 1};return versionedPtr.compare_exchange_weak(current, newValue);}return false;
}

3. 合理使用原子操作

// ✅ 适合使用原子操作的场景
std::atomic<bool> isReady{false};     // 简单标志
std::atomic<int> refCount{0};         // 引用计数
std::atomic<size_t> taskCount{0};     // 任务计数// ❌ 不适合使用原子操作的场景
// 复杂的数据结构更新应该使用锁
struct ComplexData {int value1;int value2;std::string description;
};
// 对整个结构的更新应该用mutex保护，而不是让每个字段都是原子的