2. 线程间共享数据

共享数据带来的问题

并发编程为我们带来了更高的性能和响应能力，但也引入了新的挑战，尤其是在多个线程需要访问和修改共享数据时。正确地管理这些共享数据对于确保程序的正确性和稳定性至关重要。

条件竞争 (Race Condition)

这一节是理解并发编程为何困难的基石。它主要阐述了两个核心概念：不变量 (Invariants) 的破坏是问题的表现形式，而 数据竞争 (Data Race) 是C++标准中导致这种问题的底层原因，并会引发未定义行为 (Undefined Behavior)。

不变量 (Invariants) 的破坏和条件竞争

首先，我们来理解什么是“不变量”以及它为何如此重要。

通俗理解, 不变量是数据结构在任何“稳定”时刻都必须遵守的规则。它就像一个数据结构的“健康承诺”, 确保数据结构在任何时候都处于一个有效的状态。

例如, 一个计数器变量 count 必须始终等于列表中实际的项数; 在一个双向链表中，如果节点A的“下一个”指针（next）指向节点B，那么节点B的“前一个”指针（prev）必须指向节点A。

但是不变量是如何被破坏的呢？

问题在于，几乎所有的数据结构更新操作都不是原子的（即一步完成）。它们需要多个步骤，而在这些步骤的中间，不变量是暂时被破坏的。

让我们来看一个完美阐释不变量被破坏的双向链表删除节点的例子。

不变量：如上所述，A->next == B 和 B->prev == A 必须同时成立。
删除操作的步骤（假设删除节点N，其前一个节点是P，后一个节点是S）：
- 找到要删除的节点N。
- 更新前一个节点P的 next 指针，使其指向N的下一个节点S（即 P->next = S）。
- 更新后一个节点S的 prev 指针，使其指向N的前一个节点P（即 S->prev = P）。
- 释放节点N的内存。
不变量被破坏的时刻：在第2步和第3步之间, 此时，P->next 已经指向了 S（P->next = S），但 S 的 prev 指针仍然指向 N（S->prev == N）。在这个瞬间，数据结构处于一个“非健康”或“损坏”的状态。P 和 S 之间的双向链接是不一致的。
并发访问的后果：如果此时（第2步和第3步之间），线程B开始从 P 节点向后遍历链表，它会通过 P->next 直接跳到 S，跳过了节点N（N此时理论上还在链表中）。
- 更糟糕的是，如果线程C试图从 S 节点向前遍历，它会访问到 S->prev，也就是 N，但 N 的前一个节点 P 已经不指向 N 了。
- 如果线程D试图删除节点 S，它在修改 P 节点时可能会与线程A（正在删除N）发生冲突，导致链表被永久性损坏，最终导致程序崩溃。

上面描述的这种“后果取决于哪个线程先执行”的现象，就是条件竞争。

并发中竞争条件的形成，取决于一个以上线程的相对执行顺序，每个线程都抢着完成自己的任务。

当然，并不是所有的竞争条件都会导致严重的问题。我们可以将竞争条件分为两类：

良性竞争 (Benign)：两个线程都往一个队列里添加任务。谁先谁后可能无所谓，队列的“不变量”（如队列的完整性）没有被破坏。
恶性竞争 (Problematic)：这就是我们真正关心的。它特指那些在不变量被破坏时发生的竞争，如上面的双向链表示例。

条件竞争的特点, 也是我们面对条件竞争时的痛点：

难以复现：它发生的概率可能很低，只在极端的线程调度顺序下才出现。
受负载影响：系统负载越大，线程切换越频繁，问题复现的概率越高。
调试时消失：在调试模式下，程序的执行时序会发生变化（例如I/O变慢、断点），导致竞争条件“神秘消失”，这使得它极难被定位。

数据竞争 (Data Race) 与未定义行为

C++标准库为上述提到的这种混乱状态提供了一个精确的定义:

数据竞争 (Data Race): 当两个或以上的线程并发（非同步）地访问同一个内存地址，并且至少有一个访问是写操作时，即构成数据竞争。(是条件竞争的一种具体表现形式)

在双向链表的例子中，当线程A在写入 P->next 指针时，如果线程B正在读取 P->next（用于遍历），这就构成了一次数据竞争。

更重要的是, 数据竞争是未定义行为 (Undefined Behavior) 的一种原因。

“未定义行为”是C++标准中的“最高警告”。它意味着程序不再受C++标准的任何约束, 且编译器可以（并且经常会）进行“激进”的优化，它会假设你的代码中没有数据竞争。

当数据竞争实际发生时，结果可能是：程序崩溃、数据损坏、计算出错误结果，或者（最坏的情况下）程序看起来“正常工作”了很长一段时间，直到在某个关键时刻才爆发。

避免恶性条件竞争

在上一节我们明确了问题的根源：多线程修改共享数据时，会导致数据结构的不变量 (Invariants) 被暂时破坏，如果此时被其他线程访问，就会产生恶性条件竞争。

本小节则宏观地提出了三种解决这个问题的策略。其核心思想是：必须采用一种方法，确保当一个线程正在修改数据（即处于破坏不变量的“中间状态”）时，其他线程无法访问这些数据。从其他线程的视角来看，这个修改操作必须是原子的——要么是“已经完成了”，要么是“还没开始”。

保护机制（互斥量）

这是最简单、最直接的办法，也是本章后续内容的核心。

它的核心思想：对数据结构采用某种保护机制（例如一把“锁”）。

线程A在开始修改数据之前，先“锁住”这个数据。
线程A执行修改操作（此时不变量可能被暂时破坏，但因为数据被锁住了，所以很安全）。
线程A完成修改，恢复不变量。
线程A“解锁”数据。

效果：当线程A持有锁时，任何其他试图访问该数据的线程（如线程B）都必须等待，直到线程A释放锁。这样就保证了线程B永远不会看到数据结构在“中间状态”时的样子。

C++标准库提供了很多此类机制，而最基本、最通用的就是 互斥量 (Mutex), 也就是我们后面要重点讲解的内容。

无锁编程 (Lock-free programming)

这是一种完全不同的、更高级也更复杂的思路。它不使用“锁”，而是从根本上改变数据结构的设计。

核心思想是：重新设计数据结构及其不变量，使得每一次修改都能通过一系列不可分割的变化来完成。

例如, 修改操作（添加或删除节点等）被设计成一个或多个原子操作 (Atomic Operations)。在任何一个步骤中，数据结构的不变量都保持稳定（或者说，它能“原子地”从一个稳定状态跳转到另一个稳定状态，没有中间态）。

由于不变量始终保持稳定，其他线程可以随时安全地访问数据结构，即使有线程正在对其进行修改，也不需要等待。

不过, 无锁编程对于程序员来说是一个巨大的挑战。它需要对C++内存模型（第5章）有深刻的理解，是一种专家级的技术（第7章）。

软件事务内存 (STM)

这是一种借鉴自数据库领域的、偏理论和研究的策略。

核心思想是：像操作数据库一样，使用事务 (Transaction) 来处理数据更新。

线程A开始一个“事务”。
线程A读取所需数据，并将所有修改操作记录在“事务日志”中（此时真正的共享数据还没被修改）。
线程A完成所有操作后，尝试“提交 (Commit)”这个事务，将日志中的所有修改合为一步，一次性应用到共享数据上。

如果在线程A执行事务的过程中（从第1步到第3步），有其他线程（线程B）已经修改了A所依赖的数据，那么线程A的“提交”就会失败。如果提交失败，事务将回滚 (Rollback)，并且线程A必须重启整个操作。

这保证了所有修改要么全部成功（原子提交），要么全部失败（回滚），数据结构永远不会停留在不一致的状态。

这是一个热门的研究领域，但在C++标准中没有直接支持，因此本书不会深入探讨。

使用互斥量保护共享数据

这一节是上面提出的“保护机制”策略的具体C++实现。它介绍了用于保护共享数据的最核心工具：std::mutex，以及使用它的最佳实践：std::lock_guard, 和更高级的 std::unique_lock。

C++ 中使用互斥量

std::mutex：基本的锁机制

头文件：std::mutex 和 std::lock_guard 都在 <mutex> 头文件中声明。
创建：通过实例化 std::mutex 来创建一个互斥量对象。

1 2	#include <mutex> std::mutex some_mutex;

核心操作（不推荐直接使用）：

lock()：上锁。如果互斥量 some_mutex 没有被任何线程持有，则调用此函数的线程将获得锁，并继续执行。如果锁已被其他线程持有，则调用 lock() 的线程将被阻塞 (block)，即暂停执行，直到该锁被释放。
unlock()：解锁。释放该线程对互斥量的所有权，以便其他正在等待的线程可以获取它。

声明互斥量之后, 就意味着 mutex 可以保护由 lock() 和 unlock()（或 RAII 封装如 std::lock_guard）界定的临界区（Critical Section）, 在这里线程可以安全地访问和修改共享数据。

特定的互斥量实例 some_mutex 只能保护与其关联的共享数据, 确保在任何时刻, 只有一个线程可以进入由 some_mutex 保护的临界区。多个互斥量可以保护不同的数据。

理解互斥量

互斥量的本质是一个同步原语（Synchronization Primitive），它在底层依赖于原子操作和操作系统（OS）内核来实现排他性。

核心机制：原子操作（Hardware Support）

互斥量的实现离不开 CPU 提供的原子指令，这是实现“锁”的基础，例如 Test-and-Set、Compare-and-Swap (CAS) 或 Exchange。

当一个线程尝试获取锁时，它会执行一个原子指令来检查锁的状态并尝试将其设置为“已锁定”。原子性保证了检查和设置这两个操作是不可分割的。

如果锁未被占用，原子操作成功，线程获得锁。
如果锁已被占用，原子操作失败，线程进入等待状态。

这个过程确保了两个线程不可能同时成功地将锁的状态从“未锁定”改为“已锁定”。

接着就是等待和阻塞（OS Kernel Support）

当锁已被占用，线程不能简单地空转等待（空转等待被称为忙等待/自旋锁，会浪费 CPU 资源）。高效的互斥量会依赖操作系统将等待的线程挂起（Suspend）。

即线程阻塞：当一个线程无法获取互斥量时，它会通过调用操作系统内核的函数将自己置于阻塞状态（Blocking）。

接着是内核调度：操作系统内核将该线程从 CPU 的调度队列中移除，并将其放入该互斥量关联的等待队列中。此时，该线程不再消耗 CPU 资源。

最后是线程唤醒：当持有锁的线程调用 unlock() 时，操作系统内核会被通知。内核会从等待队列中选择一个或多个线程，将其唤醒（Wake up）并重新放入 CPU 调度队列，使其有机会竞争该锁。

另外, 互斥量除了提供排他性之外，还提供了至关重要的内存同步功能（或称内存可见性）, 这是通过对lock() 和 unlock() 的内存屏障 (Memory Barriers) 实现的。

unlock() 的作用（释放）：确保在 unlock() 之前对共享内存所做的所有写入操作，对所有其他线程都是可见的。这通常涉及清空或刷新 CPU 缓存。
lock() 的作用（获取）：确保在 lock() 之后，线程能够看到先前持有该锁的线程所做的所有修改。

使用 std::lock_guard (RAII) 确保安全

虽然可以直接使用 std::mutex 的 lock() 和 unlock() 方法来保护临界区，但这要求你必须“记住在每个函数出口都要去调用 unlock()”。

这很容易出错，尤其是在函数中途有多个返回路径，或者在持锁期间发生异常时。

多重返回路径：如果你的函数很复杂，在中间有 return 语句，你可能会忘记在 return 之前调用 unlock()。
异常 (Exceptions)：如果在 lock() 和 unlock() 之间发生了异常，unlock() 调用将被跳过。

如果 unlock() 没有被及时调用，互斥量将永远保持锁定状态。任何其他试图 lock() 这个互斥量的线程都将永久阻塞，导致整个程序死锁或挂起。

为了解决上述危险，C++标准库提供了 std::lock_guard，这是一个遵循 RAII（Resource Acquisition Is Initialization，资源获取即初始化, 将资源的生命周期与一个对象的生命周期绑定）惯用语法的模板类。

std::lock_guard 的工作原理：

构造：std::lock_guard<std::mutex> guard(some_mutex);. 当你创建 guard 对象时，它会在其构造函数中自动调用 some_mutex.lock()。
- 如果 some_mutex 已被其他线程锁定，构造 guard 的这一行代码将会阻塞，直到锁可用。
析构：当 guard 对象离开其作用域时，guard 的析构函数会被调用, 且其析构函数还会自动调用 some_mutex.unlock()。

好处：

绝对安全：无论函数是正常结束、从中间 return，还是因为抛出异常而退出，guard 对象都会被正确析构。
保证解锁：这意味着互斥量总是会被正确地解锁。你再也不用手动管理 unlock() 了。

下面是一个使用 std::lock_guard 的示例：

#include <list>
#include <mutex>
#include <algorithm>

std::list<int> some_list;   // 1. 全局共享数据
std::mutex some_mutex;      // 2. 用于保护 some_list 的全局互斥量

void add_to_list(int new_value)
{
  // 3. 创建 guard 对象，自动调用 some_mutex.lock()
  std::lock_guard<std::mutex> guard(some_mutex);  
  some_list.push_back(new_value);
} // guard 在此离开作用域，自动调用 some_mutex.unlock()

bool list_contains(int value_to_find)
{
  // 4. 创建 guard 对象，自动调用 some_mutex.lock()
  std::lock_guard<std::mutex> guard(some_mutex);  
  return std::find(some_list.begin(), some_list.end(), value_to_find) 
         != some_list.end();
} // guard 在此离开作用域，自动调用 some_mutex.unlock()

add_to_list() 是一个“写者”函数，list_contains() 是一个“读者”函数。它们都试图访问同一个共享资源 some_list, 且它们都使用了同一个互斥量 some_mutex 来进行保护。

互斥量和共享资源应该是一一对应的关系。每个共享资源都应该有一个专门的互斥量来保护它，避免不同资源之间的锁冲突。

由于 std::lock_guard 的存在，系统保证在任何时刻，只有一个线程可以执行 add_to_list() 或 list_contains() 中被 guard 保护的代码块。

如果线程A正在执行 add_to_list()（已持有锁），线程B试图调用 list_contains()，线程B将在 std::lock_guard 的构造函数处被阻塞，直到线程A退出 add_to_list() 并释放锁。

结果：list_contains() 永远不可能看到 add_to_list() 正在修改列表的“中间状态”。它看到的列表要么是修改前的，要么是修改后的。不变量得到了保护。

将互斥量和数据封装在类中

虽然上述示例能工作，但使用全局变量（some_list）和全局互斥量（some_mutex）通常是糟糕的面向对象设计。

更好的做法是将数据和用于保护它的互斥量封装在同一个类中, 同时在所有公开的需要访问该数据的成员函数中使用 std::lock_guard。

class protected_list {
private:
    std::list<int> some_list;
    std::mutex some_mutex; // 互斥量和数据都是 private

public:
    void add_to_list(int new_value) {
        std::lock_guard<std::mutex> guard(some_mutex);
        some_list.push_back(new_value);
    }

    bool list_contains(int value_to_find) {
        std::lock_guard<std::mutex> guard(some_mutex);
        // ... (find logic) ...
    }
};

优势：

封装：将数据和锁紧密联系在一起。
清晰：private 访问修饰符使得所有人都清楚地知道这些数据是受保护的。
强制安全：外部代码无法直接访问 some_list 或 some_mutex，它们必须通过 protected_list 的公共成员函数。只要所有公共成员函数都正确使用了 std::lock_guard，数据访问就是绝对安全的。

精心组织代码来保护共享数据

尽管使用互斥量和 std::lock_guard 可以保护共享数据，但仅仅这样做是不够的。你必须精心设计和组织代码，以避免一些常见的陷阱和错误。

（陷阱）避免返回受保护数据的指针或引用

在设计类的接口时要格外小心，以防止受保护数据的指针或引用“泄漏”出去

std::lock_guard 的保护是有作用域的。它只能保护在 lock_guard 对象存在期间（即锁被持有的期间）执行的代码。

如果一个函数（即使它内部正确地使用了 std::lock_guard）以任何方式将一个指向“被保护数据”的原始指针或引用交给了外部代码，那么：

std::lock_guard 会在函数返回时释放锁。
外部代码现在拥有了一个指向数据的“后门”
外部代码可以在不获取锁的情况下，通过这个“后门”指针/引用随时访问甚至修改数据。

此时，如果另一个线程正在（合法地）持有锁并修改数据，而外部代码通过“后门”也在访问数据，数据竞争 (Data Race) 就发生了。

下面是通过返回值或输出参数实现的泄露, 这是最明显的一种泄漏：

class BadDesign {
private:
    std::mutex m;
    int protected_data;
public:
    // 危险：返回了受保护数据的引用！
    int& get_data() {
        std::lock_guard<std::mutex> guard(m);
        return protected_data; 
    }
};

BadDesign bd;
int& leaked_ref = bd.get_data(); // 锁在 get_data() 返回时已释放
leaked_ref = 10; // 灾难：在无保护的情况下修改了数据！

get_data() 函数本身是线程安全的，但它返回的引用 leaked_ref 却成了一个“迷失的引用”。任何持有 leaked_ref 的代码都可以绕过互斥锁 m 来读写 protected_data。

（陷阱）避免在持锁时调用用户代码

还有一种方式是通过传递给用户提供的函数泄露, 这是一种更隐蔽、更危险的泄漏方式。

// 要保护的数据
class some_data
{
  int a;
  std::string b;
public:
  void do_something();
};

// 封装数据的类
class data_wrapper
{
private:
  some_data data;
  std::mutex m;
public:
  template<typename Function>
  void process_data(Function func)
  {
    std::lock_guard<std::mutex> l(m);
    func(data);  // 1. 漏洞所在：将“受保护”数据传递给用户函数
  }
};

// --- 恶意用户的代码 ---

some_data* unprotected; // 一个全局指针，用于“接应”泄漏的数据

void malicious_function(some_data& protected_data)
{
  unprotected = &protected_data; // 将数据的地址存储到全局指针中
}

data_wrapper x;

void foo()
{
  // 2. 传递恶意函数，执行“攻击”
  x.process_data(malicious_function); 

  // 3. 在无保护的情况下访问保护数据！
  // 此时，`process_data` 已返回，锁已释放
  // 但 `unprotected` 指针已指向 `x` 内部的 `data`
  unprotected->do_something(); 
}

data_wrapper::process_data 函数看起来是安全的。它正确地用 std::lock_guard 锁定了互斥量 m。

但是, 在持有锁的同时，它调用了用户传入的函数 func，并将受保护的 data 的引用作为参数传了进去。

这就给了用户代码(传入的函数)一个机会，可以在函数执行期间(也是持锁期间) 偷偷保存 data 的地址，从而绕过互斥量的保护。

process_data 函数返回，std::lock_guard 析构，锁被释放。但是此时用户代码仍然可以通过全局指针 unprotected 直接调用 data 的成员函数，完全绕过了互斥量 m 的保护。

如果此时有另一个线程（线程B）正在调用 x.process_data(…)（并持有了锁），foo 函数（线程A）对 unprotected->do_something() 的调用就会与线程B的操作发生数据竞争，导致未定义行为。

通过上述两个陷阱，我们可以看到，即使正确使用了互斥量和 std::lock_guard，设计不当的接口仍然可能导致保护机制失效，从而引发数据竞争。

保护是必须的，但接口设计决定了保护是否有效。

C++线程库无法帮你自动防止这种逻辑错误。程序员必须遵守一个严格的准则：“切勿将受保护数据的指针或引用传递到互斥锁作用域之外”

这包括：

不作为函数返回值。
不存储在外部可见的内存中（如全局变量、或传递给你的其他对象的成员）。
不作为参数传递到用户提供的（你无法控制的）函数中去。

如果需要让用户操作数据，最好的实现是拷贝数据的副本传给用户函数，或者提供一个受保护的接口，让用户通过这个接口间接操作数据，而不是直接访问数据本身。

发现接口内在的条件竞争

在实际的编程中, 即使你完美地用 std::lock_guard 保护了每个函数, 没有泄露任何指针或引用，你的类依然可能存在严重的条件竞争。

问题不再是实现的错误，而是接口设计本身就“内置”了竞争, 各个操作之间是有“间隙”的

std::lock_guard 保证了单个成员函数（如 push()）的执行是原子的——它要么没开始，要么就做完了。

但是，如果用户为了完成一个逻辑操作（比如“获取并删除栈顶元素”），需要调用两个或更多的、各自独立的、原子的成员函数时，问题就出现了。

在第一个函数调用（持有锁A，释放锁A）和第二个函数调用（持有锁B，释放锁B）之间，存在一个“间隙”。在这个“间隙”中，本线程不持有锁，其他线程可以自由进入并修改数据结构，从而破坏本线程的操作逻辑。

std::stack 的 empty()/top() 示例 (TOCTTOU 竞争)

假设我们有一个 threadsafe_stack，它的每个成员函数（empty, top, pop）内部都正确使用了 std::lock_guard。

在单线程中，我们经常这样写：

// 单线程安全代码
stack<int> s;
if (!s.empty()) {  // 1. 检查 (Check)
  int const value = s.top(); // 2. 使用 (Use)
  s.pop();
  do_something(value);
}

这段代码依赖于一个假设：如果在第1行 empty() 返回 false，那么到第2行 top() 执行时，栈仍然是非空的。

在多线程环境下，这个假设完全失效：

线程A 调用 s.empty()。s.empty() 内部加锁、检查、返回 false（假设栈内有1个元素）、释放锁。
（间隙）此时，操作系统切换到线程B。
线程B 调用 s.empty()（返回 false）、s.top()（获取元素）、s.pop()（栈变空了）。线程B的锁被释放。
（间隙）操作系统切换回线程A。
线程A 执行到第2行 s.top()。s.top() 内部加锁，但此时栈已经是空的！
结果：对空栈调用 top() 是未定义行为 (Undefined Behavior)，程序很可能崩溃。

这就是一个经典的 TOCTTOU（Time-of-Check to Time-of-Use，检查时-使用时）条件竞争。这个竞争的产生，不是因为 empty() 或 top() 没有加锁，而是因为接口设计迫使你将“检查”和“使用”分成了两个独立的操作。

std::stack 的 top()/pop() 示例 (重复处理)

类似地，假设我们想要“获取并删除栈顶元素”，我们可能会写如下代码：

// 假设栈 s 中有两个元素 [B, A] (A在栈顶)
// 线程A 和 线程B 同时执行以下逻辑：
if (!s.empty()) {
  int const value = s.top(); // 2
  s.pop();                 // 3
  do_something(value);
}

一种可能的执行顺序：

线程A：s.empty() 返回 false。（锁释放）
线程B：s.empty() 返回 false。（锁释放）
线程A：s.top()。value 被赋值为 A。（锁释放）
线程B：s.top()。value 也被赋值为 A（因为线程A还没 pop）。（锁释放）
线程A：s.pop()。栈 s 移除 A，现在栈顶是 B。（锁释放）
线程A：do_something(A)。
线程B：s.pop()。栈 s 移除 B，现在栈是空的。（锁释放）
线程B：do_something(A)。

结果：元素 A 被处理了两次，而元素 B 被 pop 掉后永远地丢失了。这不是程序崩溃，而是一个更难排查的逻辑错误。

异常安全对接口设计的影响

既然问题出在接口上，解决方案就是改变接口。必须将多个分离的步骤合并成一个单一的、原子的操作。

我们希望的逻辑是“获取并删除栈顶元素”。std::stack 将其分为 top() 和 pop() 两个独立的操作，这就导致了上述的条件竞争。

你可能会想：“为什么 std::stack 不直接提供一个 T pop() 函数来返回值呢？”

答案是异常安全 (Exception Safety)。

想象一下 stack<std::vector<int>> s;. 如果我们调用 std::vector<int> value = s.pop();, pop() 函数的实现可能是：

从栈中移除 vector（此时已从栈上分离）。
返回 vector（这会触发拷贝构造函数）。
std::vector 的拷贝构造函数需要分配内存，如果内存不足，它会抛出 std::bad_alloc 异常。

后果：栈顶元素在第1步被成功移除了，但在第2步拷贝时失败了。这个数据永久丢失了！

除了这里之外, 在return语句中返回值的拷贝构造也可能抛出异常。

std::stack 将 top() 和 pop() 分开，就是为了让你能安全地先 top() 获取引用，安全地拷贝它（如果拷贝失败，栈本身没变），确认拷贝成功后，然后才调用 pop() 删除它。

设计线程安全的 threadsafe_stack

如果要优化上述问题, 我们的新接口必须同时满足：

线程安全：top 和 pop 的逻辑必须是原子的，不能有“间隙”。
异常安全：在返回值的过程中如果发生异常，数据不能丢失。

解决方案1: 提供一个新的成员函数 bool try_pop(T& value)，它将“获取并删除栈顶元素”的逻辑合并在一起：

template<typename T>
class threadsafe_stack {
private:
    std::stack<T> data_stack;
    std::mutex m;
public:
    bool try_pop(T& value) {
        std::lock_guard<std::mutex> guard(m);
        if (data_stack.empty())
            return false; // 栈空，无法弹出
        value = data_stack.top(); // 获取栈顶元素的副本
        data_stack.pop();         // 删除栈顶元素
        return true;              // 成功弹出
    }
    // 其他成员函数 (push, empty, etc.)
};

优点是可以保证没有数据冲突, 如果第1步的赋值操作 value = data.top() 抛出异常（例如 T 的赋值运算符抛异常），data.pop() 不会被执行，栈保持不变，数据不丢失。同时不通过 return 返回实际的值，而是通过引用参数传递，避免了返回值拷贝可能抛异常的问题。

缺点是要求 T 类型必须支持赋值，并且用户需要先构造一个 T 的实例传进去，可能有额外开销。

解决方案2: 使用智能指针 (如 std::shared_ptr<T>) 来避免拷贝：

std::shared_ptr<T> pop() {
    std::lock_guard<std::mutex> lock(m);
    if(data.empty()) throw empty_stack();
    // 1. 在栈顶元素被 pop 之前，为其创建一个 shared_ptr
    std::shared_ptr<T> const res(std::make_shared<T>(data.top())); 
    data.pop(); // 2. pop 栈
    return res; // 3. 返回 shared_ptr
}

这是解决异常安全问题的绝佳方案。第1步 make_shared 可能会因内存不足而抛异常，但此时 data.pop() 也没执行，栈保持不变。第3步返回 shared_ptr 本身（拷贝指针）是不会抛出异常的。

缺点就是, 对于 int 这样的简单类型，使用 shared_ptr 会带来不必要的堆分配和管理开销。

死锁：问题描述及解决方案

在前面的小节中，我们主要担心的是“保护不足”或“接口竞争”。而本节介绍了一个完全相反的问题：过度保护或保护顺序不当导致的“永久等待”，即死锁 (Deadlock)。

死锁的定义（两个线程互相等待）

死锁，也称为“致命拥抱”(Deadly Embrace)，发生在两个或多个线程互相等待对方释放资源（锁）时。由于所有线程都在等待，没有线程可以继续执行来释放它持有的锁，因此所有相关线程都将永久阻塞。

当一个操作需要同时获取两个或更多的互斥量时，死锁的风险就出现了。

一个常见的错误建议是：“只要所有线程总是以相同的顺序获取锁（例如，总是先锁A，再锁B），就不会死锁。”, 但这个建议在实践中很难遵守。例如下面的一个类的 swap 函数，用于交换两个实例的内容。为了保证交换的原子性，你必须同时锁住两个实例。

class X {
private:
    some_big_object data;
    std::mutex m;
public:
    friend void swap(X& lhs, X& rhs);
};

使用 std::lock() 一次性锁定多个互斥量

为了解决这种“需要一次性获取多个锁”的困境，C++标准库提供了 std::lock 函数。

std::lock (在 <mutex> 中)是一个可变参数模板函数，可以接受任意数量的“可锁定”对象（如 std::mutex）: std::lock(m1, m2, m3, ...);, 它会以一种避免死锁的算法来锁定所有传入的互斥量。

C++标准库保证：std::lock 返回时，你要么成功获取了所有的锁，要么一个都没获取到（例如，如果它在尝试获取锁的途中抛出异常，它会保证释放掉已经获取的锁）。它绝不会只持有一部分锁而导致死锁。

但是, std::lock 只负责“上锁”，它不负责“解锁”。如果我们手动 unlock()，又会遇到忘记 unlock 或异常安全的问题。因此，最佳实践是将 std::lock 与 std::lock_guard 结合使用，但需要一个特殊的技巧：使用 std::adopt_lock 标志。

std::adopt_lock 的使用

std::adopt_lock 是一个标志，表示“我已经拥有了这个锁”，用于告诉 std::lock_guard 不要尝试再次锁定互斥量。下面是一个使用 std::lock 和 std::adopt_lock 的示例：

// 清单 3.6: 交换操作中使用 std::lock() 和 std::lock_guard
class X
{
// ... (同上) ...
public:
  friend void swap(X& lhs, X& rhs)
  {
    // 1. 检查是否为同一实例
    if(&lhs == &rhs)
      return;

    // 2. 核心：无死锁地锁定两个互斥量
    std::lock(lhs.m, rhs.m); 

    // 3. 创建 lock_guard，并传入 std::adopt_lock
    //    告诉 lock_a "你不需要再 lock() 了，
    //    你只需要领养(adopt)这个已经锁定的互斥量，并在析构时 unlock() 它"
    std::lock_guard<std::mutex> lock_a(lhs.m, std::adopt_lock); // 标记 2

    // 4. 对 rhs.m 做同样的操作
    std::lock_guard<std::mutex> lock_b(rhs.m, std::adopt_lock); // 标记 3

    // 5. 现在可以安全地执行操作了
    swap(lhs.some_detail, rhs.some_detail);
  } // lock_b 和 lock_a 在此析构，自动调用 unlock()
};

if(&lhs==&rhs) 是极端重要的: 同一个 std::mutex（非递归互斥量）在同一个线程上被 lock() 两次是未定义行为。如果用户调用 swap(my_x, my_x)，std::lock(my_x.m, my_x.m) 就会触发未定义行为。这个检查避免了这种情况。

std::lock(lhs.m, rhs.m) 是避免死锁的关键。std::lock 内部会使用一种算法（例如，可能尝试锁定，如果失败就全部释放再重试，或者按地址排序锁定）来保证它能同时获取 lhs.m 和 rhs.m，而不会和另一个 swap(rhs, lhs) 调用的 std::lock 产生死锁。

std::adopt_lock 是一个“标签”常量。它告诉 std::lock_guard 的构造函数：“不要调用 m.lock()（因为 std::lock 已经做过了），请你“领养”这个锁的所有权，你唯一的职责就是在你被析构时调用 m.unlock()。”

这完美地将 std::lock 的死锁安全上锁功能与 std::lock_guard 的 RAII 自动解锁功能结合了起来。

避免死锁的进阶指导

std::lock() 是一个很棒的工具，但它只解决了“同时获取多个锁”这一个特定问题。而死锁是一个更广泛的系统设计问题。本节提供了四个超越 std::lock() 的核心设计准则，以及一个扩展思考，帮助你从架构层面根除死锁。

(1) 避免嵌套锁

这是最简单、最激进的规则。如果一个线程在任何时候最多只持有一个锁，那么“循环等待”的条件就永远无法形成，死锁也就不可能发生。因为死锁的最小条件是：线程A持有锁1，等待锁2; 且线程B持有锁2，等待锁1。

如果遵循“避免嵌套锁”规则，线程A在持有锁1时，根本不允许它再去尝试获取锁2。它必须先释放锁1，才能去获取锁2。这就打破了死锁的第一个条件。

在实践中, 当你发现你需要持有锁A的同时去获取锁B时，重新审视你的设计。是否可以先释放A？或者，是否可以将两个锁合并为一个？如果实在无法避免，请使用准则三或准则四。

(2) 避免在持有锁时调用用户代码

“用户提供的代码”是指你无法控制的代码，例如回调函数、虚函数、模板参数的函数、Lambda 表达式等。

这条准则是准则一（避免嵌套锁）的一个重要推论。你（调用者）可能没有在持有锁时获取第二个锁，但你调用的“用户代码”可能会去获取第二个锁（甚至是第一个锁，导致递归死锁）。

隐蔽的嵌套锁：你的代码（线程A）持有 lock_A，然后调用 user_function()。user_function() 内部尝试获取 lock_B。
经典的死锁：与此同时，线程B持有 lock_B，调用了另一个函数，这个函数尝试获取 lock_A。死锁发生。

在实践中, 在调用用户代码之前释放你的锁。如果用户代码需要访问受保护的数据，请先在锁内将数据复制一份，然后在锁外将副本传递给用户代码。

(3) 使用固定顺序获取锁

核心思想：如果“避免嵌套锁”不可行，你必须获取多个锁，那么强制所有线程在任何时候都必须以完全相同的、全局固定的顺序来获取这些锁。

这是最经典的死锁预防算法。我们给所有互斥量一个全局唯一的排序（例如，按它们的内存地址排序，或按功能命名排序）。

假设全局顺序是 lock_A -> lock_B, 线程A想获取A和B，它先锁A，再锁B。线程B想获取A和B，它也必须先锁A，再锁B。

如果线程A获得了 lock_A，正在等待 lock_B。线程B此时不可能持有 lock_B 并等待 lock_A。为什么？因为它必须先获取 lock_A 才能去获取 lock_B，而 lock_A 已经被线程A持有了。线程B要么在等待 lock_A（被A阻塞），要么还没开始。

这种方式将“循环等待”变成了“单向排队”。

(4) 使用锁的层次结构（hierarchical_mutex 示例）

这是准则三（固定顺序）的一个更高级、更灵活、可运行时检查的实现, 为系统中每一个互斥量分配一个唯一的“层级值”（一个数字）。高层锁有高值（如 10000），低层锁有低值（如 5000）。

一个线程在已经持有某个层级的锁时，只能再去获取比它层级更低（数字更小）的锁, 从而保证锁的获取顺序总是“从高到低”。

这种“只准向下”的规则使得“循环等待”在逻辑上变得不可能。

线程A：获取 lock_10000 (高) -> 获取 lock_5000 (低)。（允许）
线程B：获取 lock_5000 (低) -> 尝试获取 lock_10000 (高)。（禁止！）

线程B的非法尝试会被立即在运行时检测到并抛出异常，而不是引发一个难以复现的死锁。

hierarchical_mutex high_level_mutex(10000); // 1. 层级为 10000
hierarchical_mutex low_level_mutex(5000);  // 2. 层级为 5000

// ...

void low_level_func()
{
  std::lock_guard<hierarchical_mutex> lk(low_level_mutex); // 3. 锁 5000
  // ...
}

void high_level_func()
{
  std::lock_guard<hierarchical_mutex> lk(high_level_mutex); // 4. 锁 10000
  high_level_stuff(low_level_func()); // 5. 在持有 10000 时，调用函数去锁 5000
}

void thread_a()  // 6. 遵守规则的线程
{
  high_level_func();
}

hierarchical_mutex other_mutex(100); // 7. 另一个层级为 100 的锁

void other_stuff()
{
  high_level_func();  // 8. 尝试在持有 100 时，调用函数去锁 10000
  do_other_stuff();
}

void thread_b() // 9. 违反规则的线程
{
  std::lock_guard<hierarchical_mutex> lk(other_mutex); // 10. 锁 100
  other_stuff();
}

上面就是一个使用 hierarchical_mutex 的示例, thread_a 遵守了层级规则，而 thread_b 违反了规则: 在 thread_b 中，other_stuff() 试图在持有层级为 100 的锁时调用 high_level_func()，而 high_level_func() 试图获取层级为 10000 的锁。这违反了“只能获取更低层级锁”的规则，因此 hierarchical_mutex 会在运行时抛出一个异常，防止死锁的发生。

这里实现层级锁的关键在于 thread_local 变量 current_hierarchy_value，它记录了当前线程持有的最高层级锁的层级值。每次获取一个新的锁时，都会检查这个值，确保新锁的层级低于当前持有的锁。

class hierarchical_mutex
{
  std::mutex internal_mutex;
  unsigned long const hierarchy_value; // 这个互斥量实例的层级值
  unsigned long previous_hierarchy_value;

  // 1. 核心：每个线程独有一份的“当前层级”变量
  static thread_local unsigned long this_thread_hierarchy_value; 

  void check_for_hierarchy_violation()
  {
    // 2. 规则检查
    if(this_thread_hierarchy_value <= hierarchy_value) 
    {
      throw std::logic_error(“mutex hierarchy violated”);
    }
  }

  void update_hierarchy_value()
  {
    // 3. “下降”：保存旧层级，设置新层级
    previous_hierarchy_value=this_thread_hierarchy_value; 
    this_thread_hierarchy_value=hierarchy_value;
  }

public:
  explicit hierarchical_mutex(unsigned long value):
      hierarchy_value(value), previous_hierarchy_value(0)
  {}

  void lock()
  {
    check_for_hierarchy_violation(); // 检查
    internal_mutex.lock();           // 4. 锁住真正的互斥量
    update_hierarchy_value();        // 5. 更新本线程的层级
  }

  void unlock()
  {
    // 6. “上升”：恢复到上一个层级
    this_thread_hierarchy_value=previous_hierarchy_value; 
    internal_mutex.unlock();
  }
  // ... try_lock() 类似 ...
};

// 8. (在类外) 初始化线程本地变量
thread_local unsigned long
    hierarchical_mutex::this_thread_hierarchy_value(ULONG_MAX);

static 意味着它属于类，而不是类的实例; thread_local 意味着每个线程都拥有它自己独立的一个副本。

因此，this_thread_hierarchy_value 完美地跟踪了当前线程所持有的最高层级锁（准确地说是最后一个锁）的层级。它被初始化为 ULONG_MAX ，一个极大的值，表示“最高层级”，因此任何锁在开始时都可以被获取。

std::unique_lock——灵活的锁

在之前，我们学习了 std::lock_guard。它是一个严格的 RAII 包装器：一旦创建，它就必须锁定互斥量；一旦销毁，它就必须解锁互斥量。它非常高效，但也非常“死板”。

本节引入了一个更强大、更灵活的工具：std::unique_lock。

与 std::lock_guard 的对比

std::lock_guard：始终拥有它所管理的互斥量。它不维护任何状态，因此体积小、速度快。

而std::unique_lock：不一定拥有它所管理的互斥量。它内部维护一个状态标志 (flag) 来跟踪自己当前是否拥有锁。

这种灵活性是 std::unique_lock 一切功能的来源，但它也带来了轻微的代价：

空间代价：它需要存储这个状态标志，因此 std::unique_lock 对象通常比 std::lock_guard 对象更大。
时间代价：在构造、析构和所有操作中，它都需要检查或更新这个标志，因此会比 std::lock_guard 稍慢一点。

std::defer_lock 的使用（延迟加锁）

std::unique_lock 的灵活性首先体现在构造函数上。std::lock_guard 只有一种构造方式（立即锁定），而 std::unique_lock 有多种，其中最重要的是 std::defer_lock, 这是一个传递给构造函数的“标签”常量。

它告诉 std::unique_lock 的构造函数：“请不要在构造时锁定互斥量。你只管关联这个互斥量，但保持你的‘拥有锁’状态标志为 false。”. 这样做是为了能将“上锁”这个动作推迟到以后执行，特别是为了能将 std::unique_lock 对象本身传递给 std::lock() 函数。

// 清单 3.9: 交换操作中 std::lock() 和 std::unique_lock 的使用
class X
{
private:
  some_big_object some_detail;
  std::mutex m;
public:
  X(some_big_object const& sd):some_detail(sd){}
  friend void swap(X& lhs, X& rhs)
  {
    if(&lhs==&rhs)
      return;

    // 1. 延迟锁定：创建 lock_a 和 lock_b，但它们暂不持有锁
    std::unique_lock<std::mutex> lock_a(lhs.m, std::defer_lock); 
    std::unique_lock<std::mutex> lock_b(rhs.m, std::defer_lock); 

    // 2. 互斥量在这里被（无死锁地）锁定
    std::lock(lock_a, lock_b); 
    
    swap(lhs.some_detail, rhs.some_detail);
  } // lock_a 和 lock_b 在此析构，并自动解锁
};

在上面的代码中, 我们首先创建了两个 std::unique_lock 对象 lock_a 和 lock_b，并传入 std::defer_lock 标志，表示它们暂时不持有锁, 这里的延迟锁定是为了能将 lock_a 和 lock_b 传递给 std::lock() 函数, 从而实现无死锁地一次性锁定两个互斥量。(这比使用 std::lock_guard 更加灵活, 因为它不需要 std::adopt_lock 这个额外的“标签”)

当然, 除了 std::defer_lock 实现延迟锁定, std::unique_lock 还有更加灵活的用法: - 所有权转移：锁可以像对象一样被移动（move） - 手动解锁和重新加锁：可以在持有锁期间临时释放锁，然后再重新获取锁 - 条件变量的集成：std::unique_lock 是与条件变量 (std::condition_variable) 一起使用的标准锁类型，因为它允许在等待条件时释放锁。

不过, 代价就是 std::unique_lock 体积更大，速度稍慢。

不同域中互斥量所有权的传递

这一节的核心是 std::unique_lock 的一个强大特性，它来源于C++11的移动语义 (Move Semantics)。

std::unique_lock 的移动语义 (Move Semantics)

首先, 锁可以是“可移动”的，但一定是“不可拷贝”的。

你不能“拷贝”一个锁的所有权。std::lock_guard 和 std::unique_lock 都禁止拷贝。这在逻辑上是说得通的：锁是“独占”的。你不能把一个锁复制一份，然后让两个人同时“拥有”这个锁。

1 2	std::unique_lock<std::mutex> lk1(m); std::unique_lock<std::mutex> lk2 = lk1; // 编译错误！

不过, std::unique_lock 是可移动的。“移动”意味着所有权的转移。它不像拷贝（“我有一份，你也有一份”），而是像转交（“我把它给你了，我就没有了”）。

1 2	std::unique_lock<std::mutex> lk1(m); std::unique_lock<std::mutex> lk2 = std::move(lk1); // 正确！

执行后, lk2 现在拥有对互斥量 m 的锁。它负责在将来解锁 m。lk1 不再拥有锁。它仍然是一个 std::unique_lock 对象，但其内部的状态标志为 false（不拥有锁）。当 lk1 析构时，它不会去解锁 m。

从函数返回锁

这种“可移动”的特性，最常见的应用场景就是跨函数（跨作用域）传递锁的所有权。

第一种情况是隐式移动：当 std::unique_lock 作为函数返回值时，C++编译器会自动为你调用移动构造函数，你不需要显式使用 std::move()。

std::unique_lock<std::mutex> get_lock()
{
  extern std::mutex some_mutex;  // 假设在某处定义了这个互斥量
  std::unique_lock<std::mutex> lk(some_mutex); // lk 在此获取了锁
  prepare_data(); // 在持有锁的情况下准备数据
  
  return lk;  // 1. 隐式移动发生
}

void process_data()
{
  // 2. 所有权被转移到这里的 lk
  std::unique_lock<std::mutex> lk(get_lock()); 

  do_something(); // 3. 在仍然持有锁的情况下处理数据
  
} // 4. lk 在此析构，释放了最初在 get_lock() 中获取的锁

我们看 get_lock 中的 return lk;: lk 是一个局部变量。当它被 return 时，编译器知道它即将被销毁，于是自动将其视为一个“右值”(rvalue)。std::unique_lock 的移动构造函数被调用，创建了一个临时的、匿名的返回值对象，这个临时对象接管了 lk 对 some_mutex 的锁所有权, 然后这个临时对象被传递给 process_data 函数中的 lk 用作移动构造。(发生了两次所有权转移)

现在, process_data 中的 lk 合法地拥有了那个最初在 get_lock 函数内部获取的锁。do_something() 在锁的保护下安全执行。process_data 结束，lk 析构，其内部标志为 true，于是它最终调用 some_mutex.unlock()，锁被正确释放。

get_lock 函数返回，其局部的 lk 对象被析构。由于所有权已“移走”，lk 的析构函数不会去解锁 some_mutex。

或许可能会好奇, 为什么要传递锁? 不能直接在 get_lock 内部完成所有工作吗? 答案是: 有时你需要在不同的作用域中持有锁, 例如在调用栈的不同层次, 或者分离不同的模块, 例如 prepare_data() 在数据层, do_something() 在业务层, 从而清晰划分职责。同时, 还可以避免重复代码 (如果有多个函数都需要以相同的方式启动一个被锁定的操作prepare_data()), 实现可插拔的业务逻辑。

除了直接将 std::unique_lock 作为返回值外，还有一种更高级、更封装的模式，称为“网关类”。其核心思想是：不直接返回 std::unique_lock（这暴露了实现细节），而是返回一个自定义的“网关”对象, 它封装了 std::unique_lock，并在析构时自动释放锁。

class LockedDataAccess {
private:
    std::unique_lock<std::mutex> lock; // 锁是它的成员
    ProtectedData& data_ref;
public:
    // 构造函数是可移动的
    LockedDataAccess(std::unique_lock<std::mutex> lk, ProtectedData& data)
        : lock(std::move(lk)), data_ref(data) {}

    // 提供安全访问数据的函数
    void write() { data_ref.do_write(); }
    int read() { return data_ref.do_read(); }

    // Gateway 对象本身也是可移动的
};

LockedDataAccess get_locked_data() {
    std::unique_lock<std::mutex> lk(some_mutex);
    // ...
    return LockedDataAccess(std::move(lk), my_data);
}

void use_it() {
    LockedDataAccess gateway = get_locked_data();
    gateway.write();
} // gateway 在此析构，其成员 lock 随之析构，释放锁

锁的粒度

这是在使用互斥量时需要权衡的一个核心设计问题。它不是一个有“唯一正确答案”的问题，而是一个关于性能和安全之间取舍的工程决策。

细粒度锁 (Fine-grained) vs. 粗粒度锁 (Coarse-grained)

首先, 锁的粒度 (Lock Granularity) 是一个描述“一个锁到底保护了多少数据”的术语。我们通常将锁的粒度分为两类：粗粒度锁 和 细粒度锁。

粗粒度锁 (Coarse-grained lock)：

定义：用一个互斥量保护大量的数据。
示例：一个全局互斥量保护程序中的所有共享数据；或者一个类的互斥量保护该类的所有成员变量。
优点：
- 简单：易于实现和推理，很难“漏掉”保护。
- 安全：对于需要访问多个数据块的复杂操作，因为所有东西都被一个锁罩住，所以天生就是安全的。
缺点：
- 性能极差 (严重瓶颈)：这会扼杀并发性。如果线程A想访问数据块1，线程B想访问完全不相关的数据块100，线程B也必须排队等待线程A释放那个唯一的锁。
  - 早期的Linux内核使用一个全局锁，导致双核系统的性能甚至不如两个单核系统。

细粒度锁 (Fine-grained lock)：

定义：使用多个互斥量，每个互斥量只保护一小部分数据。
示例：一个DNS缓存（一个 std::map），不锁住整个 map，而是为 map 中的每一条DNS记录分配一个单独的互斥量。
优点：
- 并发性高：线程A访问记录1和线程B访问记录2可以完全并行执行，因为它们获取的是不同的锁。
缺点：
- 复杂：如果一个操作需要同时访问记录1和记录2（例如，swap 操作），你就必须同时获取两个锁，这立刻带来了死锁的风险（必须使用 3.2.4 中的 std::lock）。
- 开销：更多的互斥量对象会占用更多内存。

一个关于链表的不同粒度锁: - 细粒度锁可以为链表中的每个节点分配一个互斥量，这样线程在访问不同节点时就可以并行执行。 - 但是, 如果一个操作需要遍历整个链表（例如，搜索一个值），它必须依次获取每个节点的锁，这会导致复杂性和死锁风险增加。 - 而粗粒度锁则是为整个链表分配一个互斥量，任何线程在访问链表时都必须获取这个锁，导致并发性差。

除了“锁多少数据”，粒度问题还包括“锁多长时间”。这是本节强调的另一个关键点。黄金准则是：在任何情况下，持有锁的时间应尽可能缩减到最小。

并且, 不要在持有锁的同时执行任何可能阻塞或耗时的操作。例如文件 I/O、网络请求、数据库查询、等待用户输入、sleep()，甚至尝试获取另一个锁（除非你用了 std::lock 或有层次结构）。

使用 unique_lock::unlock() 和 lock() 临时释放锁

std::lock_guard 无法解决“持有时间”的问题，因为它从构造到析构必须一直持有锁。

而 std::unique_lock 就是为此而生的。它允许你手动、临时地释放和重新获取锁。

void get_and_process_data()
{
  // 1. 获取锁，进入受保护区域
  std::unique_lock<std::mutex> my_lock(the_mutex);
  
  // 2. 快速操作：从共享结构中复制/移动出所需数据
  some_class data_to_process = get_next_data_chunk();
  
  // 3. 关键！在慢速操作之前，手动释放锁
  my_lock.unlock(); 
  
  // 4. 耗时操作：在锁之外执行。
  //    此时，其他线程可以自由地获取 a_mutex 来 get_next_data_chunk()
  result_type result = process(data_to_process);
  
  // 5. 关键！为写入结果，重新获取锁
  //    这里会阻塞，直到锁可用
  my_lock.lock(); 
  
  // 6. 快速操作：将结果写回共享结构
  write_result(data_to_process, result);

} // 7. my_lock 在析构时，检查到自己仍持有锁，于是自动解锁

这个模式是 std::unique_lock 的一个核心用途，它完美地平衡了数据保护（在第2步和第6步）和并发性能（在第4步）。

细粒度锁的陷阱：微妙的语义变化

然而, 当你试图将锁的粒度（特别是持有时间）降到最小时，你可能会在不经意间改变操作的含义。

下面是一个operator==的细粒度锁优化实现：

class Y{
private:
  int some_detail;
  mutable std::mutex m;
  int get_detail() const
  {
    std::lock_guard<std::mutex> lock_a(m);  // 1. 只在获取时加锁
    return some_detail;
  } // 锁在这里释放

public:
  Y(int sd):some_detail(sd){}

  friend bool operator==(Y const& lhs, Y const& rhs)
  {
    if(&lhs == &rhs) return true;

    int const lhs_value = lhs.get_detail();  // 2. 获取 lhs (锁A，释放A)
    int const rhs_value = rhs.get_detail();  // 3. 获取 rhs (锁B，释放B)

    return lhs_value == rhs_value; // 4. 比较
  }
};

优点（表面上的）：

锁的持有时间极短（只在 get_detail 内部）。
一次只持有一个锁，绝对不会死锁。

缺点（致命的逻辑错误）：

这个函数没有回答“lhs 和 rhs 在同一时刻是否相等？”
它回答的是“lhs 在时间点T1 的值是否等于 rhs 在时间点T2 的值？”

导致错误的条件竞争（逻辑竞争）：

初始状态：lhs.some_detail = 10, rhs.some_detail = 20。
线程A 执行到 lhs.get_detail()，得到 lhs_value 为 10。
(上下文切换)
线程B（另一个线程）执行 lhs.some_detail = 20; 和 rhs.some_detail = 10;。
(上下文切换)
线程A 执行到 rhs.get_detail()，得到 rhs_value 为 10。
线程A 执行 return lhs_value == rhs_value; (④)，即 return 10 == 10;。

结果：函数返回 true，但 lhs 和 rhs 在任何时刻都不曾相等过。

如果你真的需要原子性的比较（即比较它们在同一时刻的值），你必须同时锁住两者

friend bool operator==(Y const& lhs, Y const& rhs)
{
    if(&lhs == &rhs) return true;

    std::lock(lhs.m, rhs.m); // 同时锁住两者
    std::lock_guard<std::mutex> lock_a(lhs.m, std::adopt_lock);
    std::lock_guard<std::mutex> lock_b(rhs.m, std::adopt_lock);

    return lhs.some_detail == rhs.some_detail; // 在持有两个锁时比较
}

这个实现是逻辑正确的，但它的粒度更粗（同时持有两个锁，且持有时间更长）。

总之, “锁的粒度”是一个没有完美答案的设计权衡：

太粗：安全，但性能差。
太细：性能好，但复杂（易死锁，易出现上述的逻辑错误）。
持有时间太长：性能差（尤其是在I/O时）。
持有时间太短：可能导致逻辑错误。

你的工作是根据具体场景，选择一个“合适”的粒度。如果 std::mutex 无法在性能和安全之间提供一个“合适”的平衡（例如，读多写少的情况），你就需要下一节中介绍的替代设施。

保护共享数据的替代设施

保护共享数据的初始化过程

在上一节中，我们讨论的互斥量（std::mutex）是一种通用的保护机制，适用于数据会被反复读写的场景。

然而，有一种非常特殊但常见的场景：数据在初始化后就几乎不再改变（通常是只读的）。例如，全局配置、单例对象、数据库连接等。

一般来说, 上述这样的资源（如数据库连接、全局配置）创建开销很大，我们不想在程序启动时就创建它，而是希望在第一次使用它时才创建。

问题是：在多线程环境下，如何安全地处理这个“第一次”？

本节就探讨了解决这个问题的错误方法和正确方法。

延迟初始化 (Lazy initialization)

延迟初始化是一种设计模式，指的是推迟对象的创建或资源的分配，直到它们真正被需要的时候。

在单线程代码中，这很简单：

std::shared_ptr<some_resource> resource_ptr;
void foo()
{
  if(!resource_ptr) // 检查是否已初始化
  {
    resource_ptr.reset(new some_resource); // 1. 未初始化，则初始化
  }
  resource_ptr->do_something();
}

在并发中, 我们可能会使用 std::mutex 这样尝试

std::shared_ptr<some_resource> resource_ptr;
std::mutex resource_mutex;

void foo()
{
  std::unique_lock<std::mutex> lk(resource_mutex); // 1. 每次调用都加锁
  if(!resource_ptr)
  {
    resource_ptr.reset(new some_resource); 
  }
  lk.unlock();
  resource_ptr->do_something();
}

这样做功能上是正确的，线程安全。

但是缺点是性能极差。资源只需要初始化一次，但这段代码迫使每一个调用 foo() 的线程（即使是第1000个线程）都必须排队、获取互斥锁，仅仅是为了执行一次 if(!resource_ptr) 检查。这在初始化完成后造成了完全不必要的序列化和性能瓶颈。

（陷阱）双重检查锁定 (Double-Checked Locking) 的危害

为了避免每次都加锁，一开始程序员们发明了一种叫做“双重检查锁定, DCLP”的模式：

// 警告：这是未定义行为！
void undefined_behaviour_with_double_checked_locking()
{
  if(!resource_ptr)  // 1. 第一次检查 (无锁)
  {
    std::lock_guard<std::mutex> lk(resource_mutex);
    if(!resource_ptr)  // 2. 第二次检查 (有锁)
    {
      resource_ptr.reset(new some_resource);  // 3. 写入
    }
  }
  resource_ptr->do_something();  // 4. 使用
}

这样的设计思路其实很自然：在“快路径”（资源已初始化）上，线程看到 resource_ptr 非空，就直接跳过判断去使用，完全避免了加锁，性能极高。只有“慢路径”（第一次初始化）才需要加锁。

然而, 它是绝对错误的 : 这是一个数据竞争 (Data Race)。无锁的读取与有锁的写入之间没有同步。

致命缺陷在于, CPU或编译器可能会进行指令重排。写入操作 resource_ptr.reset(new some_resource) 包含两个步骤： - A. 分配内存并构造 some_resource 对象； - B. 将 resource_ptr 指向这块内存。

假设线程B 执行到// 3.写入时，CPU/编译器可能先执行 B (写指针)，再执行 A (构造对象)。此时 resource_ptr 已经非空，但它指向的内存上的 some_resource 对象尚未构造完成！

上下文切换后, 另一个线程A 执行到// 1.第一次检查 (无锁)。它看到了一个非空的 resource_ptr（因为线程B执行了B步骤）。线程A跳过锁，直接执行 resource_ptr->do_something()。

结果是：线程A在一个未构造完成的、半成品的对象上调用了成员函数，导致程序崩溃或数据损坏。

为了解决上述问题, C++11 提出了两种可靠的解决方案。

解决方案 1：std::call_once 和 std::once_flag

这是C++标准库提供的、专门用于“只执行一次”场景的工具。

std::once_flag：一个特殊的（不可拷贝、不可移动的）对象，用于存储“是否已执行过”的状态。
std::call_once(flag, function, …args)：一个函数，它保证 function 在多线程环境下绝对只被执行一次。

std::shared_ptr<some_resource> resource_ptr;
std::once_flag resource_flag;  // 1. 创建状态标志

void init_resource() // 初始化的函数
{
  resource_ptr.reset(new some_resource);
}

void foo()
{
  // 2. 保证 init_resource() 只被调用一次
  std::call_once(resource_flag, init_resource); 
  resource_ptr->do_something();
}

第一个调用 std::call_once 的线程会执行 init_resource 并设置 resource_flag。

其他同时调用 std::call_once 的线程会阻塞，直到 init_resource 执行完毕。

在此之后所有调用 std::call_once 的线程（例如第1000个线程）会看到 resource_flag 已被设置，于是立即返回，几乎没有开销。

这完美解决了“尝试1”的性能瓶颈，且完全线程安全。

并且, std::call_once 同样适用于类的成员变量函数（延迟初始化数据库连接）。

class X
{
private:
  connection_handle connection;
  std::once_flag connection_init_flag;

  void open_connection() { /* ... */ }
public:
  void send_data(data_packet const& data) // 1
  {
    // 2. 调用成员函数，需传入 this 指针
    std::call_once(connection_init_flag, &X::open_connection, this); 
    connection.send_data(data);
  }
  // ... (receive_data 类似) ...
};

无论是 send_data 还是 receive_data 被首先调用，std::call_once 都会保证 open_connection 只被执行一次。

解决方案 2：线程安全的 static 局部变量初始化

对于“我只需要一个全局实例”的场景，C++11提供了一个更简单、更优雅的语法, 也就是 C++11 标准的保证：

C++11强制要求 static 局部变量的初始化必须是线程安全的。

class my_class;
my_class& get_my_class_instance()
{
  // 编译器会自动保证这里的初始化是线程安全的
  // (通常内部实现就是用的 call_once)
  static my_class instance; 
  return instance;
}

当多个线程同时第一次调用 get_my_class_instance() 时，C++运行时会保证只有一个线程会执行 my_class 的构造函数，其他线程会等待。

初始化完成后，所有后续调用都会直接返回 instance 的引用。

这是实现线程安全“单例模式” (Singleton) 的最简洁、最推荐的方法。

保护很少更新的数据结构

继上面讨论了“只初始化一次”的场景后，本节讨论了另一种常见的特殊场景：“读多写少” (Read-Mostly)。

想象一个数据结构（如DNS缓存、系统配置、字典等），它需要被大量的线程频繁地读取，但只会被极少数的线程偶尔地写入（更新）。

如果使用 std::mutex, 它提供的是独占访问 (Exclusive Access)。

这意味着，如果线程A正在读取数据（持有了锁），线程B也只想读取数据（这对数据安全没有任何威胁），线程B也必须排队等待。

在“读多写少”的场景下，所有的“读者”线程都会被迫序列化（排队），这严重扼杀了并发性，使得多线程的读取性能和单线程一样差，甚至更糟（因为锁的开销）。

读者-写者锁 (Reader-Writer Mutex)

为了解决这个特定问题，我们需要一种更“聪明”的锁，它能区分“读者”和“写者”：

核心思想：

允许多个读者线程同时、并发地访问数据。
只允许一个写者线程独占地访问数据。

工作规则：

规则1 (读-读并发)：如果一个线程A持有了“读锁”，其他线程B、C、D也可以立即获得“读锁”。
规则2 (写-写互斥)：如果一个线程A持有了“写锁”，其他线程（无论是读者还是写者）都必须等待。
规则3 (读-写互斥)：如果任何线程持有了“读锁”（哪怕只有一个），一个试图获取“写锁”的线程必须等待，直到所有读者都释放了锁。

std::shared_mutex (C++17)

虽然 C++11 标准库中没有提供这种互斥量，但从 C++17 开始，标准库提供了 std::shared_mutex, 它实现了上述的读者-写者锁机制。而 std::shared_lock 则是与之配套的读锁 RAII 包装器。

头文件：std::shared_mutex 和 std::shared_lock 都在 <shared_mutex> 头文件中声明。
互斥量本身：std::shared_mutex
RAII 包装器（读者）：std::shared_lock<std::shared_mutex>, 这是 std::lock_guard 的“读者版本”。
- 它的构造函数获取一个共享锁 (Shared Lock) 或称读锁 (Read Lock)。
RAII 包装器（写者）：std::lock_guard<std::shared_mutex> 或 std::unique_lock<std::shared_mutex>
- 它们的构造函数获取一个独占锁 (Exclusive Lock) 或称写锁 (Write Lock)。

#include <map>
#include <string>
#include <mutex>
#include <shared_mutex> // 引入 C++17 的 <shared_mutex>

class dns_entry;

class dns_cache
{
  std::map<std::string, dns_entry> entries;
  mutable std::shared_mutex entry_mutex; // 使用 std::shared_mutex
public:
  // "读者" 函数 (高频)
  dns_entry find_entry(std::string const& domain) const
  {
    // 1. 获取一个“共享锁” (读锁)
    std::shared_lock<std::shared_mutex> lk(entry_mutex); 
    
    std::map<std::string, dns_entry>::const_iterator const it =
        entries.find(domain);
    return (it == entries.end()) ? dns_entry() : it->second;
  } // lk 析构，释放“共享锁”

  // "写者" 函数 (低频)
  void update_or_add_entry(std::string const& domain,
                           dns_entry const& dns_details)
  {
    // 2. 获取一个“独占锁” (写锁)
    std::lock_guard<std::shared_mutex> lk(entry_mutex); 
    
    entries[domain] = dns_details;
  } // lk 析构，释放“独占锁”
};

在上面的代码中, dns_cache 类使用 std::shared_mutex 来保护其内部的 entries 映射。如果多个线程同时调用 find_entry()，它们可以并发地获取共享锁，从而实现高效的读取。

不过, 需要注意的是, std::shared_mutex 比 std::mutex 更复杂，管理锁状态（有多少读者？有没有写者在等？）的内部开销更大。

因此是否能提升性能，完全取决于实际负载。

如果“写”操作的比例稍微高一点，或者处理器核心数较少，那么读者-写者锁的额外开销可能会抵消并发读取带来的好处，甚至可能慢于简单的 std::mutex。

在使用它之前，最好在目标系统上进行性能分析 (Profiling)，以确保它真的带来了好处。

嵌套锁

这一节讨论的是一个在面向对象编程中很常见的问题：当一个已经持有锁的函数，又调用了同一个类中的另一个也需要锁的函数时，会发生什么？

问题的根源来自 std::mutex 的“非递归”性。

std::mutex 是一个非递归互斥量。这意味着，如果一个线程已经持有了某个 std::mutex 的锁，它绝对不能尝试第二次 lock() 这个互斥量, 这样做会导致未定义行为，在很多平台上程序会立即死锁或崩溃。

假如你有一个类，它用一个 std::mutex m 来保护其所有成员数据。你遵循了之前的建议，让每个 public 成员函数都在函数开头使用 std::lock_guard 来加锁。

问题来了：假设 public 成员函数 func1() 在其实现中，需要调用另一个 public 成员函数 func2()。

class MyClass {
private:
    std::mutex m;
    int data;

public:
    void func2() {
        std::lock_guard<std::mutex> lock(m); // 2. func2 尝试第二次加锁
        // ... (操作 data)
    }

    void func1() {
        std::lock_guard<std::mutex> lock(m); // 1. func1 第一次加锁
        // ... (操作 data)

        func2(); // 3. 调用 func2

        // ...
    } // 4. func1 结束，释放锁
};

MyClass obj;
obj.func1(); // 5. 线程在这里崩溃或死锁！

func1 调用 func2(), func2 的 lock_guard 再次尝试获取锁 m。由于线程已经持有 m，这个第二次 lock() 操作导致了未定义行为。

std::recursive_mutex

为了解决上述问题, C++标准库提供了std::recursive_mutex（递归互斥量）。

std::recursive_mutex 允许同一个线程多次获取同一个锁。它在内部维护一个“锁定计数器”。

lock()：如果锁未被持有，线程获取它，计数器设为1。如果锁已被本线程持有，计数器+1。如果锁被其他线程持有，则阻塞。
unlock()：计数器-1。只有当计数器归零时，这个锁才会被真正释放，其他线程才能获取它。

class MyClass_Recursive {
private:
    std::recursive_mutex m; // 使用递归互斥量

public:
    void func2() {
        std::lock_guard<std::recursive_mutex> lock(m); // 2. 成功！(计数器=2)
        // ...
    } // 3. func2 析构，计数器=1

    void func1() {
        std::lock_guard<std::recursive_mutex> lock(m); // 1. 成功！(计数器=1)
        // ...
        func2();
        // ...
    } // 4. func1 析构，计数器=0，锁被释放
};

在使用时, 我们可以简单地将 std::mutex m; 替换为 std::recursive_mutex m;

std::lock_guard（和 std::unique_lock）可以完美地配合 std::recursive_mutex 工作，它们会正确地处理计数。

（警告）嵌套锁通常是设计缺陷的标志

尽管 std::recursive_mutex 解决了燃眉之急，但强烈建议不要使用它。std::recursive_mutex 往往是糟糕设计的标志。

回想一下，我们加锁的根本原因是什么？是为了在我们修改数据时，保护数据结构，因为在修改的中间过程，类的不变量（必须保持的规则）可能会被暂时破坏。

在 func1 的例子中：

func1 加锁（计数器=1），因为它准备修改数据，不变量可能即将被破坏。
func1 修改了数据A，但还没来得及修改数据B（此时不变量已破坏）。
func1 调用了 func2。
func2 成功加锁（计数器=2），它开始在一个不变量已被破坏的对象上执行操作！

这是具有极大风险的：func2 的代码很可能是基于“类的所有不变量都保持完好”这个假设来编写的。但当它被 func1 在“中间状态”调用时，这个假设不成立了，可能导致 func2 产生错误的计算或逻辑混乱。

更好的解决方案是重构代码, 分离“加锁逻辑”和“工作逻辑”。

我们提取出一个新的私有 (private) 成员函数，这个函数不加锁，它假定调用者已经持有了锁。

class MyClass_Refactored {
private:
    std::mutex m; // 1. 换回 std::mutex
    int data;

    // 2. 新的 private 辅助函数，它不加锁！
    // 命名中的 "_impl" 或 "_unsafe" 常用
    void func2_impl() { 
        // 假设 m 已经被锁住
        // ... (操作 data)
    }

public:
    // 3. func2 的 public 版本只负责加锁
    void func2() {
        std::lock_guard<std::mutex> lock(m);
        func2_impl(); // 调用无锁的实现
    }

    void func1() {
        std::lock_guard<std::mutex> lock(m); // 4. func1 加锁
        // ... (操作 data)
        
        func2_impl(); // 5. 直接调用无锁的实现
        
        // ...
    }
};

这种设计更清晰地分离了职责：public 函数负责接口和线程安全（加锁），private 函数负责核心实现（假定已安全）。

这也确保了 func2_impl 要么被 func1 在一个完整的、原子的操作中调用，要么被 func2 在一个完整的、原子的操作中调用，不变量始终受到保护。