互斥量和原子操作解决多线程数据共享

在多个线程中共享数据时，需要注意线程安全问题。如果多个线程同时访问同一个变量，并且其中至少有一个线程对该变量进行了写操作，那么就会出现数据竞争问题。数据竞争可能会导致程序崩溃、产生未定义的结果，或者得到错误的结果。为了避免数据竞争问题，需要使用同步机制来确保多个线程之间对共享数据的访问是安全的。常见的同步机制包括互斥量、条件变量、原子操作等。

互斥量

问题的根源：竞争条件 (Race Condition)

在介绍解决方案之前，我们必须清晰地理解问题所在。当多个线程同时访问和修改同一个共享数据时，就会产生竞争条件。我们来看一个最经典的例子：多个线程同时对一个全局计数器进行递增操作。

#include <iostream>
#include <thread>
#include <vector>

int g_counter = 0; // 全局共享变量

void increment() {
    for (int i = 0; i < 100000; ++i) {
        g_counter++; // <--- 问题的核心：临界区
    }
}

int main() {
    std::vector<std::thread> threads;
    for (int i = 0; i < 10; ++i) {
        threads.push_back(std::thread(increment));
    }

    for (auto& th : threads) {
        th.join();
    }

    // 理论上，10个线程每个加10万次，结果应该是 1,000,000
    // 但实际运行结果会是一个小于一百万的随机数
    std::cout << "Final counter value: " << g_counter << std::endl; 
    
    return 0;
}

实际结果出错的原因在于, g_counter++ 这一行代码在底层并不是一个原子操作 (Atomic Operation)。它至少包含三个步骤： 1. 读取 (Read)：从内存中读取 g_counter 的当前值到一个CPU寄存器。 2. 修改 (Modify)：在CPU寄存器中将该值加 1。 3. 写入 (Write)：将寄存器中的新值写回到内存中的 g_counter。

想象一下两个线程同时执行的场景： - 线程A 读取 g_counter 的值（假设为 100）到它的寄存器。 - 在线程A修改之前, 线程B 也读取 g_counter 的值（此时内存中仍然是 100）到它的寄存器。 - 线程B 在自己的寄存器中加 1（变为 101），并将其写回内存。现在 g_counter 的值是 101。 - 由于线程运行的异步性, 此时线程A才对它自己的寄存器（值仍然是 100）加 1，得到 101。 - 线程A 将 101 写回到内存。g_counter 的值仍然是 101。

最终结果：两个线程都执行了 ++ 操作，但计数器只增加了 1。这就是数据竞争导致的最终结果不一致。这块访问共享资源的代码 g_counter++，我们称之为临界区 (Critical Section)。我们的目标就是保护它。

解决方案：互斥量 (std::mutex)

互斥量，顾名思义，就是互斥访问 (Mutual Exclusion)。它就像一把锁，用来保护一段代码（临界区）。其基本规则如下：

一个线程想要进入临界区，必须先拿到互斥量(锁)。如果锁被其他线程占用，当前线程就会阻塞等待，直到锁被释放。
如果锁可用，当前线程就会拿到锁，进入临界区执行操作。
在此期间，如果其他线程也想进入，它们会发现锁被占用，只能在外面阻塞等待，直到锁被释放。
第一个线程执行完临界区代码后，会释放锁 (unlock)，把锁放回原处。
等待的线程中会有一个拿到锁，进入临界区执行操作。

通过这种方式，我们保证了在任何时刻，只有一个线程能进入临界区。

互斥量的基本使用

首先需要引入头文件: #include <mutex>

手动调用 lock() 和 unlock()

这是最基本的使用方式，需要在临界区代码前后手动调用 lock() 和 unlock() 方法。

#include <iostream>
#include <thread>
#include <vector>
#include <mutex>

int g_counter = 0;
std::mutex g_mutex; // 创建一个全局互斥量

void safe_increment() {
    for (int i = 0; i < 100000; ++i) {
        g_mutex.lock();   // 在访问共享数据前加锁
        g_counter++;
        g_mutex.unlock(); // 在访问结束后解锁
    }
}
// ... main 函数与之前相同，只是调用 safe_increment

但这种手动管理的方式有致命缺陷：如果在 lock() 和 unlock() 之间发生异常，unlock() 就永远不会被调用，导致互斥量被永久锁定，所有其他等待该锁的线程都会被无限期阻塞，这被称为死锁 (Deadlock)。

std::lock_guard (推荐的现代方法)

为了解决手动解锁的风险，C++ 标准库提供了 std::lock_guard，它完美地利用了 RAII (Resource Acquisition Is Initialization，资源获取即初始化) 的思想。

std::lock_guard 是一个类模板，它的工作方式是： - 在构造时：它会自动接收一个 std::mutex 对象，并在构造函数调用该对象的 lock() 方法。 - 在析构时：当 lock_guard 对象离开其作用域时（例如，在代码块 {} 的末尾），它的析构函数会自动被调用，并在析构函数中调用 unlock() 方法。 - std::lock_guard对象不能复制或移动，因此它只能在局部作用域中使用。 - std::lock_guard 的职责是在其生命周期内“拥有”一个互斥锁。如果它能被复制，那么我们就会有两个 lock_guard 对象都认为自己拥有同一个锁。当这两个对象离开作用域时，它们的析构函数都会尝试去调用 unlock()。对一个已经解锁的互斥量再次解锁是未定义行为，会导致程序错误。因此，从逻辑上讲，复制 lock_guard 是不安全的，所以 C++ 禁止了这种行为。 - “移动”一个对象意味着将资源的所有权从一个对象转移到另一个对象。如果 std::lock_guard 可以被移动（例如，从一个函数返回），那么“解锁”这个行为的发生地点就不再是创建锁的那个原始、清晰的局部作用域了，而是转移到了一个不确定的新作用域。std::lock_guard 的设计目标就是简单和绝对的安全。它的理念是：“锁在哪里创建，就必须在哪里被释放，绝不允许所有权转移”。禁止移动特性，就是为了强制执行这种简单、可预测、不会出错的模式。 - 正是因为 std::lock_guard 不能被复制或移动，它的应用场景就被严格地限制在了创建它的那个局部作用域 (local scope) 内。这意味着你不能将 std::lock_guard 作为函数参数按值传递; 不能从一个函数返回一个 std::lock_guard 对象; 不能把它存入一个容器（如 std::vector）; 也不能把它作为一个类的成员变量，然后在不同实例间赋值或转移。 - 这个限制不是一个缺陷，而是一个特性。它通过牺牲灵活性来换取极致的简单和安全，杜绝了因锁的所有权混乱而导致的死锁或未定义行为。它是一个“做一件事并把它做到完美”的工具。****

#include <iostream>
#include <thread>
#include <vector>
#include <mutex>

int g_counter = 0;
std::mutex g_mutex;

void best_safe_increment() {
    for (int i = 0; i < 100000; ++i) {
        // 创建 lock_guard 对象，它在构造时自动锁住 g_mutex
        std::lock_guard<std::mutex> lock(g_mutex);
        
        g_counter++;
        
        // 当 lock 对象离开这个作用域时（for循环的本次迭代结束），
        // 它的析构函数会自动调用 g_mutex.unlock()
        // 即使 g_counter++ 抛出异常，也能保证解锁！
    }
}
// ... main 函数与之前相同，只是调用 best_safe_increment

需要注意的是, 我们应该保持临界区简短：加锁会阻塞其他线程，影响并发性能。因此，被锁定的代码块应该尽可能小，只包含必要的操作，然后尽快释放锁

std::unique_lock (推荐的现代方法)

std::unique_lock 是一个更强大、更灵活的锁管理器。它与 std::lock_guard 的核心区别在于：std::unique_lock 实现了可移动 (movable) 的所有权语义，并提供了更丰富的手动操作接口。

可移动，不可复制 (Movable, Non-copyable): std::unique_lock 像 std::unique_ptr 一样，遵循唯一所有权模型。你不能复制它，但可以移动它，从而实现锁的所有权转移。

std::unique_lock<std::mutex> create_lock() {
    std::unique_lock<std::mutex> u_lock(g_mutex);
    // ... do something ...
    return u_lock; // 所有权被转移出去 (隐式移动)
    // 此时在析构时不会解锁，因为锁的所有权已经转移
}

void another_function() {
    std::unique_lock<std::mutex> received_lock = create_lock();
    // 现在 received_lock 拥有锁，当它离开作用域时会负责解锁
}

延迟锁定 (Deferred Locking): std::lock_guard 在构造时必须锁定。而 std::unique_lock 可以选择在构造时不锁定，之后再手动锁定。

std::unique_lock<std::mutex> u_lock(g_mutex, std::defer_lock);
// 此时互斥量 g_mutex 并未被锁定

// ... 在未来的某个时刻 ...
u_lock.lock(); // 手动加锁

手动控制: std::unique_lock 允许你在其生命周期内手动调用 lock() 和 unlock()。这允许你实现更细粒度的锁定策略：在不需要锁的时候提前释放它，以提高并发性。

std::unique_lock<std::mutex> u_lock(g_mutex); // 立即锁定
// ... 执行一小部分需要锁的代码 ...

u_lock.unlock(); // 提前解锁，让其他线程可以工作

// ... 执行很长的、不需要锁的代码 ...

u_lock.lock(); // 再次加锁
// ... 执行另一部分需要锁的代码 ...

// 函数结束时，如果 u_lock 仍持有锁，RAII 机制会保证它被解锁
// 如果移动了锁的所有权，则不会解锁

与条件变量 (std::condition_variable) 配合使用: 这是 std::unique_lock 最重要的用途。条件变量的 wait() 方法要求传入一个 std::unique_lock。因为它需要在等待时原子地解锁互斥量，并在被唤醒后自动重新加锁。std::lock_guard 无法提供这种手动解锁和重新加锁的灵活性。

死锁 (Deadlock) 问题

死锁是指两个或多个线程在执行过程中，因争夺资源而造成的一种互相等待的僵局。在这种状态下，如果没有外力干预，这些线程都将无法向前推进，导致整个程序或系统的相关部分被“冻结”。而在编程中, 死锁争夺的资源就是互斥量 (mutex)。

例如下列代码

#include <iostream>
#include <thread>
#include <mutex>

std::mutex mutex_A;  // 互斥量 A
std::mutex mutex_B;  // 互斥量 B
int account_A = 1000;
int account_B = 2000;

// 线程1: 尝试从 A 转账到 B
void transfer_A_to_B() {
    mutex_A.lock(); // 1. 成功锁住 mutex_A
    std::this_thread::sleep_for(std::chrono::milliseconds(10)); // 模拟一些操作
    mutex_B.lock(); // 3. 尝试锁住 mutex_B，但它被线程2持有，于是线程1开始阻塞等待
    // ... 转账操作 ...
    mutex_A.unlock();
    mutex_B.unlock();
}

// 线程2: 尝试从 B 转账到 A
void transfer_B_to_A() {
    mutex_B.lock(); // 2. 成功锁住 mutex_B
    std::this_thread::sleep_for(std::chrono::milliseconds(10)); // 模拟一些操作
    mutex_A.lock(); // 4. 尝试锁住 mutex_A，但它被线程1持有，于是线程2开始阻塞等待
    // ... 转账操作 ...
    mutex_B.unlock();
    mutex_A.unlock();

}

int main() {
    std::thread t1(transfer_A_to_B);
    std::thread t2(transfer_B_to_A);
    t1.join();
    t2.join();

    std::cout << "All transfers finished.\n"; // 这句话可能永远不会被打印
    return 0;
}

上述示例的结局是线程1 持有 mutex_A，等待 mutex_B; 线程2 持有 mutex_B，等待 mutex_A。两个线程都将永远等待下去，程序被挂起。

死锁的四个必要条件

一个死锁的发生，必须同时满足以下四个条件（因为发表人的原因, 也被称为“Coffman条件”）： - 互斥条件 (Mutual Exclusion)：资源不能被共享，在任意时刻，一个资源只能被一个线程持有。 (互斥量天生就满足此条件) - 持有并等待条件 (Hold and Wait)：一个线程至少持有一个资源，并且正在请求其它线程持有的资源。 (例如，线程1持有A，等待B) - 不可剥夺条件 (No Preemption)：资源不能被强制地从一个线程中抢占，只能由持有它的线程自愿释放。 (你不能强制线程1释放mutex_A) - 循环等待条件 (Circular Wait)：存在一个线程的等待链，使得 P1 等待 P2 的资源，P2 等待 P3 的资源，… ，Pn 等待 P1 的资源，形成一个环路。 (我们的例子中 T1 -> T2 -> T1 就是一个环)

避免死锁的方法

避免死锁的思路就是破坏这四个条件中的至少一个, 最常见和最实用的方法是破坏“循环等待”条件。 1. 按固定顺序加锁 (最常用的方法) 这是解决死锁问题的黄金法则。规定程序中所有需要同时锁住多个互斥量的地方，都必须严格按照相同的全局顺序来获取锁。

例如，我们可以规定，总是先锁地址较小的那个互斥量。

// 假设 mutex_A 的地址 < mutex_B 的地址

void safe_transfer() { // 两个线程都调用这一个函数
    // 总是先锁地址较小的 mutex_A，再锁地址较大的 mutex_B
    std::lock_guard<std::mutex> lock_a(mutex_A); 
    std::this_thread::sleep_for(std::chrono::milliseconds(10));
    std::lock_guard<std::mutex> lock_b(mutex_B);

    // ... 执行转账操作 ...
}

在这个修正版中，无论是A到B还是B到A的转账，都会先尝试锁mutex_A，再尝试锁mutex_B。这样，当一个线程成功锁住mutex_A后，另一个线程会因为无法锁住mutex_A而直接等待，它根本没有机会去锁住mutex_B，因此循环等待的条件被破坏，死锁不会发生。

使用 std::lock (C++11/17 推荐) 手动管理锁的顺序可能很复杂且容易出错。C++标准库提供了一个完美的工具 std::lock，它可以一次性原子地锁住多个互斥量，并且内部实现了避免死锁的算法。

void best_safe_transfer() {
    // std::lock 会以一种避免死锁的方式锁住两个互斥量
    std::lock(mutex_A, mutex_B);

    // 使用 std::adopt_lock 参数告诉 lock_guard，互斥量已经被锁住，
    // 它只需要负责在析构时解锁即可。
    std::lock_guard<std::mutex> guard_a(mutex_A, std::adopt_lock);
    std::lock_guard<std::mutex> guard_b(mutex_B, std::adopt_lock);

    // ... 执行转账操作 ...
}

这是目前处理多个互斥量加锁问题的最安全、最推荐的方案。

其他策略

破坏“持有并等待”：尝试一次性获取所有需要的锁（std::lock就是这样做的），如果不能，就释放所有已持有的锁并重试。这可以使用 std::unique_lock 的 try_lock 方法实现，但逻辑更复杂。
减少锁的粒度：尽量不要长时间持有锁，尤其不要在持有锁的时候做耗时操作（如文件I/O），让临界区尽可能小。
避免嵌套锁：尽量避免在一个锁的作用域内再去获取另一个锁，如果不可避免，请严格遵守加锁顺序。

原子操作

上述属于传统的解决方案, 即是使用互斥锁 (std::mutex) 来保护共享数据，确保同一时间只有一个线程可以访问 counter。然而，互斥锁是操作系统层面的同步原语，涉及系统调用，可能会导致线程阻塞和上下文切换，开销相对较大。

std::atomic 提供了一种更轻量级、更底层的解决方案。它利用现代 CPU 提供的特殊原子指令（如 LOCK CMPXCHG），在硬件层面保证单个操作的原子性 (Atomicity)，从而避免数据竞争，且通常比互斥锁性能更高。

std::atomic<T> 是一个模板类，它包装了一个 T 类型的值，并确保对这个值的所有操作都是原子的。原子操作是指一个从所有其他线程的角度来看不可分割的操作。它要么完全执行，要么完全不执行，不存在任何中间状态被其他线程观察到。

std::atomic 定义在 <atomic> 头文件中。

基本用法与操作

#include <atomic>

std::atomic<int> atomic_counter(0);  // 声明一个原子整数并初始化为 0
std::atomic<bool> is_ready(false);  // 声明一个原子布尔值并初始化为 false

std::atomic 的成员函数可以分为几类：

写入与读取 (Store and Load)
- store(value): 原子地将 value 写入原子对象(也就是赋值)。
- load(): 原子地读取原子对象的值(也就是取值)。
- 常用的赋值和读取操作符被重载，通常会调用这两个函数。
读-修改-写 (Read-Modify-Write, RMW) 操作: 这是 std::atomic 最强大的功能，它将读取、修改、写入三个步骤合并为一个不可分割的原子操作。
- exchange(value): 原子地将原子对象的值替换为 value，并返回替换前的旧值。
- fetch_add(arg), fetch_sub(arg): 原子地给当前值加上/减去 arg，并返回操作前的旧值。重载的 ++, –, +=, -= 等操作符通常调用它们。
compare_exchange_strong(expected, desired) / compare_exchange_weak(expected, desired): 这是最核心的 RMW 操作（比较并交换，CAS）。工作流程：
- 比较原子对象的当前值与 expected 的值。
- 如果相等，则将原子对象的值修改为 desired，并返回 true。
- 如果不相等，则将 expected 的值更新为原子对象的当前值，并返回 false。 > strong vs weak: strong 保证如果值相等，交换就一定成功。weak 版本在某些平台上性能更好，但即使值相等也可能“伪失败”（spurious failure），即返回 false。因此 weak 版本通常用在循环中。

atomic_counter.store(10); // 等同于 atomic_counter = 10;
int current_val = atomic_counter.load(); // 等同于 int current_val = atomic_counter;

// 解决最开始的计数器问题
std::atomic<int> counter(0);
counter++; // 原子操作，不会产生数据竞争

std::atomic<int> val(10);
int expected = 10;
int desired = 20;
// 尝试将 val 从 10 原子地更新为 20
if (val.compare_exchange_strong(expected, desired)) {
    // 成功，val 现在是 20
} else {
    // 失败，可能是因为其他线程修改了 val
    // 此时 expected 的值会被更新为 val 的当前值
}

内存序 (Memory Ordering)

原子性仅仅保证了单个操作的不可分割性，但并未规定该操作与其他内存读写操作之间的顺序。为了性能，编译器和 CPU 可能会对指令进行重排序。在单线程中，这毫无问题。但在多线程中，这种重排可能会导致灾难性的后果。

// 共享变量
int shared_data = 0;
std::atomic<bool> data_ready = false;

// 线程 A: 生产者
void producer() {
    shared_data = 42;                   // 操作 A
    data_ready.store(true);             // 操作 B
}

// 线程 B: 消费者
void consumer() {
    if (data_ready.load()) {            // 操作 C
        assert(shared_data == 42);      // 操作 D
    }
}

从程序员的逻辑来看，producer 中 A 操作一定先于 B 操作。consumer 只有在 C 操作读到 true 之后，才会执行 D 操作。因此，assert 应该永远不会失败。

但现实是, 编译器或 CPU 可能会认为操作 A 和 B 互不依赖，为了优化，可能会将它们的执行顺序重排。producer 的实际执行顺序可能变成：

// 重排后的生产者
void producer_reordered() {
    data_ready.store(true);             // 操作 B
    shared_data = 42;                   // 操作 A
}

也就是当线程 B 执行 assert(shared_data == 42)时, shared_data 还是 0，断言失败！

内存序就是用来约束这种重排序，确保多线程间操作的可见性顺序。它的本质是一种内存屏障。内存屏障是一种指令，它告诉编译器和 CPU：“任何指令都不能跨越我这个屏障进行重排”。

std::memory_order 枚举定义了六种内存序模型：

宽松序 (std::memory_order_relaxed):最弱的内存序, 只保证当前原子操作的原子性，不提供任何额外的同步或排序保证。其他线程可能以任意顺序观察到内存的修改, 指令可以在任意时间点执行, 也可以被重排。
- 适用场景: 只关心单个原子变量的修改，不依赖它来同步其他数据。例如，简单地增加一个计数器，只在最后才读取它的值。
获取-释放语序 (Acquire-Release Semantics): 这是实现线程间同步最常用的模型，通常成对出现。
- std::memory_order_release:用于写入/存储操作。它确保在当前线程中，所有位于此 release 操作之前的内存写入（无论是原子还是非原子的），对于在其他线程中对同一个原子变量执行 acquire 操作的线程都是可见的, 换句话说, 它是一个向上的屏障, 在此 release 操作之前的所有内存写入，都不能被重排到这个操作之后(编译期)。
  - 实际上, 这个命令还有硬件层面的含义, 即将执行该指令的CPU核心（例如核心A）的缓存中，所有在此之前的写入（包括 shared_data = 42），刷新到共享内存系统中。
- std::memory_order_acquire:用于读取/加载操作。它确保在当前线程中，所有位于此 acquire 操作之后的内存读取，都能看到由其他线程中对同一个原子变量执行 release 操作的线程所写入的数据。换句话说, 它是一个向下的屏障, 在此 acquire 操作之后的所有内存读取，都不能被重排到这个操作之前。
  - 在硬件层面, 执行该指令的CPU核心（例如核心B）在加载 shared_data 的值时，先使自己的本地缓存无效 (invalidate)，然后去共享内存系统中获取最新的值。
  - 一般与 release 配合使用，形成同步点: 读取就用 acquire，写入就用 release。
- std::memory_order_acq_rel:用于读-修改-写操作，同时具备 acquire 和 release 的特性。
顺序一致性 (std::memory_order_seq_cst):最强的内存序，也是默认的内存序。它不仅提供 acquire-release 的所有保证，还额外保证所有线程都以相同的顺序观察到所有 seq_cst 操作。它在所有线程之间建立了一个单一的、全局的操作总顺序。这个最容易理解，但通常也是性能开销最大的，因为它限制了编译器和 CPU 的优化能力。

对比

特性	std::atomic	std::mutex
保护对象	单个变量 (如int, bool, 指针)	一段代码块 (临界区)，可包含多个变量和复杂逻辑
实现原理	通常是硬件级别的原子指令 (无锁)	通常是操作系统级别的内核对象 (可能涉及线程阻塞)
性能开销	较低，适合高频访问	较高，不适合高频、短小的临界区
使用场景	简单的标志位、计数器、指针等细粒度同步	保护复杂数据结构或一系列必须整体执行的操作
死锁风险	无	有 (如果加锁顺序不当)

当需要保护的是一个独立的、简单的内置类型（如标志、计数器）时，优先使用 std::atomic。

当需要保护一个复杂的数据结构（如 std::vector）或需要将一系列操作（例如，从一个 vector 中移除元素并更新大小）作为一个不可分割的事务来执行时，必须使用 std::mutex。

自旋锁和互斥锁

自旋锁特性： - 实现机制：在用户态忙等待，不断循环检测锁状态 - CPU消耗：高，即使等待也在持续消耗CPU周期 - 上下文切换：无，线程始终保持运行状态 - 响应延迟：低，锁释放后能立即获取 - 实现复杂度：相对简单，通常基于原子操作

互斥锁特性： - 实现机制：通过操作系统内核调度，线程阻塞并进入睡眠 - CPU消耗：低，等待时不消耗CPU资源 - 上下文切换：有，涉及用户态到内核态切换 - 响应延迟：高，需要唤醒睡眠线程 - 实现复杂度：相对复杂，依赖操作系统支持

性能关键因素： - 锁竞争强度：高竞争下自旋锁性能急剧下降 - 临界区大小：短临界区适合自旋锁(锁持有时间小于两次上下文切换时间)，长临界区适合互斥锁 - CPU核心数：多核系统更适合自旋锁 - 调度策略：实时系统可能偏好自旋锁

下面是一个简单的示例

#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)) {
            // 可选的优化：在重度竞争时让出CPU
            // std::this_thread::yield();
        }
    }
    
    void unlock() {
        flag.clear(std::memory_order_release);
    }
};

// 使用示例
Spinlock spinlock;
int shared_data = 0;

void spinlock_worker(int id) {
    for (int i = 0; i < 1000; ++i) {
        spinlock.lock();
        shared_data++;  // 短临界区操作
        spinlock.unlock();
    }
    std::cout << "Thread " << id << " finished" << std::endl;
}


//------------------------------------------------------------------------


//互斥锁实现
#include <mutex>
#include <thread>
#include <iostream>

std::mutex mtx;
int shared_data = 0;

void mutex_worker(int id) {
    for (int i = 0; i < 1000; ++i) {
        std::lock_guard<std::mutex> lock(mtx);
        shared_data++;  // 短临界区操作
        // 模拟较长临界区操作
        // std::this_thread::sleep_for(std::chrono::microseconds(10));
    }
    std::cout << "Thread " << id << " finished" << std::endl;
}