前面几章讲的是"怎么创建、管理、取消线程"，这一章要解决的是更核心的一个问题：当多个线程同时读写同一块内存时，怎样才能保证数据不出错。

1. 进程有保护，线程之间却没有

操作系统对"进程"这个层级是有内存保护的：一个进程一般不能随意读写另一个进程的内存空间，操作系统会通过虚拟内存机制把不同进程互相隔离开。但是同一个进程内部的多个线程，天生共享同一份内存空间，操作系统并不会阻止线程 A 去读写线程 B 正在使用的那块内存——这本来就是线程被设计出来的目的：多个线程需要方便地共享数据来协作完成任务。
正因为线程之间的内存访问完全没有操作系统层面的保护，一旦多个线程同时对同一个内存地址进行写操作，就必须由程序员自己引入同步机制，否则会出现数据竞争（Data Race），导致数据被破坏、结果不可预测。
用一个最小的例子回忆一下这种"没有保护"会带来什么后果：

#include <iostream> // 标准输入输出
#include <thread>   // std::thread
#include <vector>   // 存放多个线程对象
int shared_value = 0; // 多个线程共享的一个普通整数变量，没有任何保护措施
// 每个线程要执行的任务：把共享变量自增 100000 次
void unsafe_increment() {
    for (int i = 0; i < 100000; ++i) {
        // 这一行看起来是一步操作，但底层其实分成"读取-加一-写回"三个步骤，
        // 多个线程交叉执行这三个步骤时就会互相覆盖对方的结果
        ++shared_value;
    }
}
int main() {
    std::vector<std::thread> threads;
    // 开 4 个线程，每个线程都在争抢着修改同一个 shared_value
    for (int i = 0; i < 4; ++i) {
        threads.emplace_back(unsafe_increment);
    }
    for (auto& t : threads) {
        t.join();
    }
    // 期望结果应该是 4 * 100000 = 400000，
    // 但因为没有任何同步保护，实际运行结果往往会比这个数字小，而且每次运行结果可能都不一样
    std::cout << "实际结果: " << shared_value << std::endl;
    std::cout << "期望结果: " << 4 * 100000 << std::endl;
    return 0;
}

这段代码每次运行的输出很可能都不完全一样，而且几乎总是小于期望值 $4\times 100000=400000$。这正是这一章要着手解决的问题：怎样借助各种"锁"相关的同步工具，让这类并发访问变得安全、可预测。
编译运行方式：

g++ -std=c++17 -pthread unsafe_increment_demo.cpp -o unsafe_increment_demo
./unsafe_increment_demo

2. 这一章要讲的内容地图

这一章会按照下面的顺序，从"发现问题"到"解决问题"层层递进：

第4章 基于锁的线程同步
├── 竞态条件
│   讲清楚竞态条件到底是什么 是怎么产生的
├── 互斥 Mutual Exclusion
│   C++中用 std::mutex 实现互斥访问
├── 通用锁管理
│   怎样更安全 更方便地管理加锁解锁这件事
├── 条件变量
│   条件变量是什么 怎么和互斥量搭配使用
├── 实现一个完全同步的队列
│   综合运用 std::mutex 和 std::condition_variable
└── C++20新增的同步原语
    ├── 信号量 semaphore
    ├── 屏障 barrier
    └── 闩 latch

这些内容有一个共同点：它们都属于基于锁的同步方式，也就是靠"让线程等待、排队"来解决并发访问的问题。与之相对的还有一种完全不同的思路——**无锁（Lock-Free）**技术，通过原子操作来避免线程真的进入等待状态，这部分内容会放在下一章单独讲解，这里先不展开。

3. 各小节内容一览

---

小节主题	要解决的核心问题
竞态条件	多线程并发访问共享内存为什么会出错 出错的根本原因是什么
互斥量 std::mutex	如何保证同一时刻只有一个线程能进入某段代码
通用锁管理	如何用更安全的方式管理加锁和解锁 避免忘记解锁导致的问题
条件变量	如何让线程等待某个条件成立后再继续执行
同步队列的实现	如何把互斥量和条件变量组合起来 实现一个线程安全的生产者消费者队列
C++20新同步原语	信号量 屏障 闩 分别用来解决哪些锁之外的协调场景

---

4. 小结

这一章的主线很清晰：先通过竞态条件理解"为什么会出问题"，再依次学习 std::mutex、通用锁管理、条件变量这几件基础工具，最后把它们组合起来实现一个真正可用的同步队列，并补充 C++20 新增的几种同步原语。掌握好这些内容，是理解后续无锁编程和原子操作章节的重要基础。

深入理解竞态条件：一个计数器引发的问题

前面已经从概念上认识过竞态条件，这一节我们用一个最简单、最经典的例子——两个线程一起给同一个计数器加一，从底层一步一步拆解，搞清楚竞态条件到底是怎么发生的，为什么结果会是错的，而且每次运行结果还都不一样。

1. 先看现象：一段"看起来没问题"的代码

#include <iostream>   // 标准输入输出
#include <thread>     // std::thread
int counter = 0; // 全局变量，两个线程都会去修改它
int main() {
    // 定义一个 lambda 表达式，作为线程要执行的任务
    // 内容很简单：循环一百万次，每次都把 counter 加一
    auto func = [] {
        for (int i = 0; i < 1000000; ++i) {
            counter++; // 关键的一行，看起来平平无奇
        }
    };
    // 创建两个线程，都执行同一个func任务
    std::thread t1(func);
    std::thread t2(func);
    // 等待两个线程都执行完毕
    t1.join();
    t2.join();
    // 打印最终的计数器值
    std::cout << counter << std::endl;
    return 0;
}

这段代码的逻辑非常直白：两个线程各自把 counter 累加 100 万次，正常来说，最终 counter 应该等于：
$$1,000,000+1,000,000=2,000,000$$
但如果真的把这段代码连续运行三次，实际得到的结果可能是这样的：

第一次运行: 1056205
第二次运行: 1217311
第三次运行: 1167474

这里暴露出两个非常反常识的问题：

1. 结果本身是错的：没有一次结果等于期望的 2000000，全部都比期望值小很多。
2. 每次运行结果都不一样：同样的代码、同样的逻辑，三次运行给出了三个完全不同的数字。
   这种"结果依赖于代码执行的具体顺序，而且这个顺序每次运行都可能不一样"的现象，就是竞态条件最典型的表现。

2. 拆解 counter++ 这一行代码

要理解为什么会出问题，关键在于看清楚 counter++; 这一行代码，在 CPU 层面实际上并不是"一步完成"的操作，而是被拆分成了三个独立的步骤：

1. 读取（Load）：把内存里 counter 变量当前的值，加载到 CPU 的一个寄存器里。
2. 计算（Increment）：把寄存器里的值加一。
3. 写回（Store）：把寄存器里新的值，写回到 counter 在内存中的地址。
   用箭头简单表示这三步的顺序：
   $$\text{读取内存值}\to \text{寄存器加一}\to \text{写回内存}$$
   需要特别说明的是：不管你写的是后缀自增 counter++，还是前缀自增 ++counter，底层都会被拆成这三步，两者在竞态条件这个问题上是完全等价的，谁都不比谁更安全。
   问题的核心在于：这三个步骤不是一个不可分割的整体（不是"原子"的），线程在执行这三步的过程中，随时可能被操作系统调度器打断，切换去执行另一个线程的代码。 如果两个线程的这三步操作交叉、穿插着执行，就会导致其中一次加一的结果被"覆盖"、凭空消失。

3. 具体推演一次"丢失更新"的过程

假设 counter 当前的值是 1，现在线程1和线程2都要对它执行一次加一操作。按照下面这种特定的执行顺序，就会发生问题：

执行步骤	线程1	线程2
[1]	把counter的值(1)读入寄存器
[2]	寄存器的值加一，变成2
[3]		把counter的值(仍然是1)读入寄存器
[4]	把寄存器的值(2)写回counter
[5]		寄存器的值加一，变成2
[6]		把寄存器的值(2)写回counter

---

按照上表这个顺序，逐步看发生了什么：

• [1]：线程1把 counter 当前的值 1 读进自己的寄存器。
• [2]：线程1把寄存器里的值加一，变成 2（注意：这个 2 目前只存在于线程1的寄存器里，counter 在内存里还是 1）。
• [3]：这时候操作系统切换去执行线程2，线程2也去读取 counter 的值——但因为线程1还没有把结果写回内存，counter 在内存里此时依然是 1，所以线程2读到的也是 1。
• [4]：操作系统又切回线程1，线程1把自己寄存器里的 2 写回内存，counter 变成了 2。
• [5]：线程2继续执行，把自己寄存器里的值（之前读到的1）加一，变成 2。
• [6]：线程2把自己寄存器里的 2 写回内存，counter 还是 2。
  最终结果：两个线程都执行了一次"加一"，理论上 counter 应该从 1 变成 3，但实际上走完这套流程之后，counter 只变成了 2。线程1那次加一操作的成果，被线程2用"旧数据算出的结果"直接覆盖掉了，这就是所谓的丢失更新（Lost Update）。

3.1 用时序图还原这次"丢失更新"

从图上能非常直观地看出问题所在：线程2在第 [3] 步读取数据的时候，线程1的计算结果还没来得及写回内存，导致线程2用的是一个"过时"的旧值去做计算，最终把线程1的劳动成果"顶替"掉了。

3.2 为什么每次运行结果都不一样

上面推演的只是众多可能交叉顺序中的一种。实际运行程序时，两个线程各自执行 100 万次循环，每一次循环里的"读取-计算-写回"三步，具体会在什么时间点被打断、被哪个线程插队，完全取决于操作系统当时的调度情况——这本身就是不确定的，受 CPU 当前的负载、调度算法的具体实现等各种因素影响。
正因为每一次程序运行时，线程之间具体的"交叉穿插方式"都不一样，所以最终有多少次"加一"操作被覆盖丢失，也就跟着不一样，这就是为什么三次运行会得到三个不同的、而且都比期望值小的结果。

4. 问题的根源：这个操作不是"原子"的

把上面的分析总结成一句话就是：counter++ 这个操作不是原子操作（atomic operation）。所谓原子操作，指的是一个操作要么完整地执行完，要么完全不执行，执行过程中不可能被其他线程"插队"看到中间状态、或者在中间步骤被打断。
如果 counter++ 真的是原子的（也就是"读取、计算、写回"这三步能够作为一个整体、不可分割地执行完，中途不允许任何其他线程插队），那么不管两个线程的执行顺序如何交替，最终结果都一定会是正确的 2000000，不会有任何一次加一操作凭空消失。但现实中，一条看似简单的高级语言代码，编译后往往对应多条机器指令，"三步操作不可分割"这个前提并不会自动成立，这正是竞态条件产生的根本原因。

5. 解决思路（预告）

既然问题的根源是"多个步骤的操作不是原子的、可以被打断插队"，那么解决思路自然也就有两个方向：

1. 用锁（Lock）来控制访问：让"读取-计算-写回"这一整套操作，在同一时刻只允许一个线程执行，其他线程必须排队等待，等前一个线程完全执行完、把锁释放之后才能进去。这样即使操作本身分成好几步，也不会被其他线程打断插队。
2. 用原子操作（Atomic Operation）：直接借助硬件层面提供的、天生保证不可分割的指令，让"加一"这个操作本身就变成一个不可能被打断的原子步骤，从根源上避免竞态条件的发生。
   这两种思路分别对应接下来要学习的锁机制和原子操作，它们是解决竞态条件最核心、最常用的两大武器，后续章节会分别展开详细介绍。

小结

• counter++ 这种看似"一行代码"的操作，底层实际上被拆成了读取、计算、写回三个独立步骤，这三步合起来并不是一个不可分割的原子操作。
• 当多个线程并发执行这种非原子操作时，一个线程的计算过程可能会被另一个线程打断、插队，导致某些线程的计算结果被后来者用"过时的旧数据"覆盖，这种现象叫做丢失更新，是竞态条件最常见的表现形式。
• 竞态条件的结果具有非确定性——同样的代码，因为每次运行时线程被调度、打断的具体时机不一样，最终结果也会不一样，而且几乎不可能"恰好"得到正确答案。
• 解决竞态条件的两大核心手段是锁（Lock）和原子操作（Atomic Operation），它们的共同目标都是让共享数据的访问变得"不可被打断"，具体用法会在接下来的章节详细展开。

为什么需要互斥访问：从咖啡店的故事讲起

一、先用一个生活中的例子理解"互斥"到底是什么

互斥访问（Mutual Exclusion）是并发编程里一个非常基础的概念，说的是：多个线程或者多个进程，不能同时访问同一个共享资源——不管这个资源是一个共享变量、一段需要保护的代码（临界区），还是一个文件、一条网络连接。互斥访问是防止竞态条件的关键手段。
用一个咖啡店的场景来理解会特别直观：假设有一家小咖啡店，只有一台意式浓缩咖啡机，这台机器一次只能做一杯咖啡，所以它是店里所有咖啡师都要共用的关键资源。店里有三位咖啡师：Alice、Bob 和 Carol，他们可以轮流用这台机器，但绝对不能同时用，不然就会出乱子。
设想这样一个场景：Bob 往机器里放好了刚磨好的咖啡粉，按下按钮开始做咖啡。这时候 Alice 也想用机器，她看到机器里有咖啡粉，以为是 Bob 忘记清理的残留物，于是顺手把粉倒掉了，然后放上自己的咖啡粉，开始做她自己的咖啡。过了一会儿，Bob 回来想拿走自己做好的咖啡——结果这一杯里其实混着两个人操作的痕迹，甚至可能两个人的咖啡都没能正常做出来。Alice 一脸茫然：明明我按流程做了，怎么没有咖啡？这就是一场"灾难"，本质上跟程序里的计数器错误累加问题一模一样。
要解决这个问题，咖啡店可以任命 Carol 当"机器管理员"：Alice 和 Bob 谁想用机器，都得先问 Carol"我现在能开始做一杯新的咖啡吗"，Carol 一次只批准一个人用，用完了才轮到下一个人。这样一来，机器在任意时刻都只服务一个人，不会再出现互相干扰的情况。
用图来梳理一下这个"有管理员协调"和"没有管理员协调"的区别：

放回到程序的世界里，"Carol 这个管理员角色"对应的，就是 C++ 标准库提供的 std::mutex——一个专门用来保护共享数据、确保同一时刻只有一个线程能访问的同步工具。mutex 这个名字本身，就是 mutual exclusion（互斥访问）这两个词的缩写。

二、回到程序问题：为什么 ++counter 这一行也需要"管理员"

2.1 ++counter 看起来简单，其实分成了好几步

++counter 这一行代码，看起来是"一个不可分割的动作"，但实际上 CPU 执行的时候，会拆分成三个独立的步骤：
$$\begin{gathered} \text{步骤一：把 counter 的当前值从内存读到寄存器} \\ \text{步骤二：把寄存器里的值加 1} \\ \text{步骤三：把加完的新值写回内存} \end{gathered}$$
问题就出在这三步不是一气呵成的，中间随时可能被另一个线程插进来。设想 counter 当前的值是 $5$，两个线程 $T_1$ 和 $T_2$ 几乎同时执行 ++counter：

这就是典型的竞态条件：两个线程都以为自己"独立地"把 counter 加了 1，但因为读取和写入这两步中间被另一个线程插了队，导致其中一次自增的结果被"覆盖"、凭空消失了。如果这样的冲突在 100 万次自增里反复发生若干次，最终 counter 的值就会比理论上的正确值小一些，而且每次运行结果可能都不一样，这正是竞态条件"结果不确定"的典型表现。

2.2 用 std::mutex 来当"管理员"

std::mutex 提供了两个最基本的操作：lock() 和 unlock()。lock() 的语义是"申请拿到这把锁"，如果这把锁当前没人用，申请立刻成功；如果已经被别的线程占用了，申请的线程就会被阻塞、原地等待，一直等到持有锁的线程调用 unlock() 释放锁为止。因为同一时刻只能有一个线程真正持有这把锁，被这把锁保护起来的那一小段代码（也就是"临界区"），在任意时刻也就只会有一个线程在执行，这就从根本上杜绝了前面讲的那种"读取-修改-写入"被别的线程插队的情况。

三、完整代码：对比"不加锁"和"加锁"两种写法

下面这段代码分别用两种方式，让两个线程各自把 counter 累加 100 万次，对比最终结果的差异。

#include <iostream>
#include <mutex>   // 提供 std::mutex
#include <thread>
std::mutex mtx;    // 保护 counter 的互斥量，全局唯一的一把"锁"
int counter = 0;   // 两个线程都会去修改的共享变量
int main() {
    // 关键点：这个 Lambda 完全没有做任何同步保护，
    // 两个线程会直接、毫无协调地去读写同一个 counter，
    // 属于典型的"没有管理员协调"的场景。
    auto funcWithoutLocks = [] {
        for (int i = 0; i < 1000000; ++i) {
            ++counter; // 危险：这一步内部其实是"读取-加一-写回"三个动作，随时可能被另一个线程打断
        }
    };
    // 关键点：这个 Lambda 在每次自增前后，分别调用 mtx.lock() 和 mtx.unlock()，
    // 保证任意时刻只有一个线程能进入 ++counter 这一行，相当于给这一行加上了"专属通道"。
    auto funcWithLocks = [] {
        for (int i = 0; i < 1000000; ++i) {
            mtx.lock();   // 申请这把锁：如果别的线程正拿着，这里就原地等待，直到轮到自己
            ++counter;    // 拿到锁之后，这一行代码此刻只有当前线程会执行，绝对安全
            mtx.unlock(); // 用完立刻释放锁，让其他正在等待的线程有机会拿到锁继续执行
        }
    };
    // 第一组实验：不加锁，直接并发自增
    {
        counter = 0; // 每组实验前重置计数器，方便对比结果
        std::thread t1(funcWithoutLocks);
        std::thread t2(funcWithoutLocks);
        t1.join();
        t2.join();
        // 理论上两个线程各加 100 万次，总共应该是 200 万，
        // 但由于竞态条件的存在，实际打印出来的值往往会比 2000000 小
        std::cout << "Counter without using locks: " << counter << std::endl;
    }
    // 第二组实验：加锁保护，确保每次自增都是独立、完整的操作
    {
        counter = 0;
        std::thread t1(funcWithLocks);
        std::thread t2(funcWithLocks);
        t1.join();
        t2.join();
        // 因为每次自增都被互斥锁完整保护起来了，不会再发生互相覆盖的问题，
        // 无论运行多少次，这里打印出来的结果都会稳定地是 2000000
        std::cout << "Counter using locks: " << counter << std::endl;
    }
    return 0;
}

代码逐段讲解：

• std::mutex mtx;：定义了一把全局唯一的互斥锁，所有想要安全访问 counter 的代码，都要通过这把锁来协调，它扮演的正是前面故事里 Carol 那个"机器管理员"的角色。
• funcWithoutLocks：这个版本里两个线程各自跑 100 万次 ++counter，中间没有任何协调机制，两个线程随时可能在"读取旧值、计算新值、写回新值"这三步之间互相插队，导致部分自增操作的结果被覆盖丢失。
• funcWithLocks：这个版本在每次自增前调用 mtx.lock()，自增之后立刻调用 mtx.unlock()。因为这把锁"同一时刻只能被一个线程拿到"，当 t1 拿到锁准备执行 ++counter 时，t2 如果也想执行这一行，就会在 mtx.lock() 这里被卡住、原地等待，一直等到 t1 调用 unlock() 释放锁为止。这样一来，"读取-加一-写回"这三个步骤对于任何一次自增来说，都不会再被另一个线程打断，counter 每一次加 1 的操作都是完整、独立的，不会再发生互相覆盖的问题。
• 两组实验分别重置 counter 为 0，跑完之后打印结果，方便直接对比"有没有加锁"造成的差异。
  多跑几次这段程序会发现：第一组（不加锁）的结果每次运行可能都不一样，而且通常小于 200 万；第二组（加锁）的结果无论运行多少次，都稳定是精确的 2000000。这正是互斥锁发挥作用的直接证明。
  用一张对比图梳理一下两种方式在"多个线程尝试修改 counter"这件事上的行为差异：

四、这两种情况的最终结果对比

---

执行方式	理论正确结果	实际观察到的结果	原因
不加锁并发自增	2000000	通常小于 2000000，且每次运行可能不同	"读取-加一-写回"这三步之间可能被另一个线程插队，导致部分自增结果被覆盖丢失
加锁保护后再自增	2000000	稳定等于 2000000	同一时刻只有一个线程能执行 ++counter，每次自增都是完整、不被打断的操作

---

五、小结

互斥访问要解决的核心问题，是"多个执行流同时碰同一份共享数据"时可能出现的互相覆盖、结果错乱的问题。C++ 标准库提供的 std::mutex，本质上就是一把"任意时刻只能被一个线程持有"的锁，通过 lock() 申请、unlock() 释放，把一段需要保护的代码（临界区）变成"排队执行"，从而保证像 ++counter 这样看似简单、实则由多个步骤组成的操作，不会因为多个线程的交叉执行而产生错误结果。理解了这个基本原理之后，再深入去看 std::mutex 本身还提供了哪些细节能力，会更容易上手。

C++ 标准库提供的互斥量实现：六种 Mutex 类型详解

前面已经讲过"互斥"这个概念本身——保证同一时刻只有一个线程能进入某段代码。这一节我们来看 C++ 标准库具体提供了哪些互斥量相关的类，它们彼此之间的差异体现在三个维度上：访问方式（独占还是共享）、是否支持递归加锁、是否支持带超时的加锁尝试。

1. 六种互斥量类型总览

标准库一共提供了六个互斥量类：

互斥量类型	访问方式	是否支持递归加锁	是否支持超时
std::mutex	独占 同一时刻只允许1个线程持有	否	否
std::recursive_mutex	独占 同一时刻只允许1个线程持有	是	否
std::shared_mutex	1个独占 或 N个共享	否	否
std::timed_mutex	独占 同一时刻只允许1个线程持有	否	是
std::recursive_timed_mutex	独占 同一时刻只允许1个线程持有	是	是
std::shared_timed_mutex	1个独占 或 N个共享	否	是

---

用一棵分类树，把这六种类型按"三个维度是否具备"梳理得更直观一些：

互斥量家族
├── 基础独占锁
│   └── std::mutex
├── 支持递归的独占锁
│   └── std::recursive_mutex
├── 支持读写共享的锁
│   └── std::shared_mutex
├── 支持超时的独占锁
│   └── std::timed_mutex
├── 支持超时且支持递归的独占锁
│   └── std::recursive_timed_mutex
└── 支持超时且支持读写共享的锁
    └── std::shared_timed_mutex

可以看出来，这六个类其实是围绕三个"能力开关"（要不要共享访问、要不要支持递归、要不要支持超时）自由组合出来的。接下来逐个看它们各自的特点和用法。

2. std::mutex：最基础的独占锁

这是最常用、最基础的互斥量：同一时刻只能有一个线程持有它，其他线程想加锁都必须排队等待，直到持有者释放为止。前面章节其实已经用过它，这里再给一份最简洁的示例巩固一下：

#include <iostream>
#include <mutex>
#include <thread>
std::mutex g_mtx;      // 定义一个基础互斥量
int g_shared_value = 0; // 需要被保护的共享变量
void add_one() {
    // lock_guard 在构造时自动加锁，离开作用域时自动解锁
    std::lock_guard<std::mutex> lock(g_mtx);
    ++g_shared_value;
}
int main() {
    std::thread t1(add_one);
    std::thread t2(add_one);
    t1.join();
    t2.join();
    std::cout << "结果: " << g_shared_value << std::endl; // 恒等于 2
    return 0;
}

编译运行：

g++ -std=c++17 -pthread basic_mutex_demo.cpp -o basic_mutex_demo
./basic_mutex_demo

3. std::recursive_mutex：允许同一个线程重复加锁

普通的 std::mutex 有一条很重要的规则：同一个线程如果对一把已经被自己锁住的 std::mutex 再次调用 lock()，会导致死锁（自己把自己卡死，因为它并不知道"这次加锁请求"和"上次加锁"其实是同一个线程发出的）。
但在某些场景下，比如一个函数内部递归调用自己，而每次调用都需要加锁保护同一段逻辑，这时候用普通 mutex 就会立刻死锁。std::recursive_mutex 就是为了解决这个问题：同一个线程可以对它重复加锁多次，不会死锁，只是解锁的次数必须和加锁的次数完全匹配，锁才会真正被释放，好让其他线程有机会获得它。

#include <iostream>
#include <mutex>
#include <thread>
std::recursive_mutex g_rec_mtx; // 支持同一线程重复加锁的互斥量
int g_depth_count = 0;
// 一个会递归调用自己的函数，每次调用都需要先加锁
void recursive_task(int depth) {
    // 同一个线程在还没解锁的情况下再次调用 lock()，
    // 如果这里用的是 std::mutex，程序会在第二层递归时直接死锁
    std::lock_guard<std::recursive_mutex> lock(g_rec_mtx);
    ++g_depth_count;
    std::cout << "当前递归深度: " << depth << std::endl;
    if (depth < 3) {
        recursive_task(depth + 1); // 递归调用，此时锁还没被释放
    }
    // 每一层的 lock_guard 在这一层函数返回时自动解锁，
    // 递归锁内部维护了一个"加锁次数"计数器，只有当计数器归零，锁才真正被释放
}
int main() {
    std::thread t(recursive_task, 1);
    t.join();
    std::cout << "总共执行了 " << g_depth_count << " 层" << std::endl;
    return 0;
}

关键点说明：

• recursive_mutex 内部维护了一个"这把锁被同一线程加了几次"的计数器，每次同一线程调用 lock() 计数器加一，每次调用 unlock() 计数器减一，只有减到 0 的时候，其他线程才真正有机会拿到这把锁。
• 递归锁虽然方便，但通常被认为是一种"能不用就不用"的工具：频繁需要递归加锁往往说明代码结构可以重新设计，用递归锁掩盖问题有时会让潜在的逻辑混乱变得更难发现。
  编译运行：

g++ -std=c++17 -pthread recursive_mutex_demo.cpp -o recursive_mutex_demo
./recursive_mutex_demo

4. std::shared_mutex：一写多读的共享锁

前面两种锁都是"独占"的：不管是读还是写，同一时刻永远只能有一个线程进去。但现实中有一类很常见的场景——很多线程只是"读"某份数据，读操作互相之间并不会造成冲突，只有"写"操作才需要真正互斥。如果这类场景还用普通的独占锁，会让本来可以并发执行的多个读操作也被迫排队，浪费性能。
std::shared_mutex 就是为这种场景设计的，它支持两种加锁模式：

• 共享模式（Shared）：多个线程可以同时以共享模式持有这把锁，适合"只读"操作，对应公式可以理解成同一时刻共享锁的持有者数量 $N$ 可以大于 1：
  $$N_{shared}\ge 0(可以同时有多个线程持有共享锁)$$
• 独占模式（Exclusive）：同一时刻只能有一个线程以独占模式持有这把锁，而且只要有人持有独占锁，其他任何线程（不管是想读还是想写）都必须等待：
  Nexclusive∈{0, 1}(独占锁的持有者数量只能是0或1) N_{exclusive} \in \{0,\ 1\} \quad (独占锁的持有者数量只能是0或1) Nexclusive​∈{0, 1}(独占锁的持有者数量只能是0或1)

#include <iostream>
#include <shared_mutex> // std::shared_mutex 定义在这个头文件里
#include <thread>
#include <vector>
#include <chrono>
std::shared_mutex g_shared_mtx; // 支持共享/独占两种模式的互斥量
int g_data = 0;                 // 被保护的共享数据
// 读线程：多个读线程可以同时进入，因为大家都是共享模式加锁
void reader_task(int id) {
    // std::shared_lock 是配合 shared_mutex 使用的共享模式加锁包装器
    std::shared_lock<std::shared_mutex> lock(g_shared_mtx);
    std::cout << "读线程 " << id << " 读到的数据: " << g_data << std::endl;
    std::this_thread::sleep_for(std::chrono::milliseconds(200)); // 模拟读取耗时
} // 离开作用域自动释放共享锁
// 写线程：写操作需要独占模式，执行期间不允许任何其他读或写线程进入
void writer_task(int new_value) {
    // std::unique_lock 是独占模式加锁包装器，效果和 lock_guard 类似，但功能更完整
    std::unique_lock<std::shared_mutex> lock(g_shared_mtx);
    g_data = new_value;
    std::cout << "写线程把数据修改为: " << new_value << std::endl;
    std::this_thread::sleep_for(std::chrono::milliseconds(200)); // 模拟写入耗时
} // 离开作用域自动释放独占锁
int main() {
    std::vector<std::thread> threads;
    // 先启动 3 个读线程，它们可以同时并发地读取数据
    for (int i = 1; i <= 3; ++i) {
        threads.emplace_back(reader_task, i);
    }
    // 再启动 1 个写线程，它必须等所有正在进行的读操作都结束后才能拿到独占锁
    threads.emplace_back(writer_task, 100);
    for (auto& t : threads) {
        t.join();
    }
    return 0;
}

关键点说明：

• std::shared_lock 对应"共享模式"加锁，多个线程可以同时持有；std::unique_lock 对应"独占模式"加锁，同一时刻只能有一个线程持有，而且它和共享模式互斥——只要有写线程在独占，读线程也进不来。
• 这种"读写锁"模式特别适合"读多写少"的场景，比如配置信息很少更新、但会被很多线程频繁查询的情况。
  编译运行：

g++ -std=c++17 -pthread shared_mutex_demo.cpp -o shared_mutex_demo
./shared_mutex_demo

5. std::timed_mutex：支持超时的独占锁

普通 std::mutex 调用 lock() 时，如果锁已经被别人占用，当前线程会无限期地阻塞等待，直到拿到锁为止。但有些场景我们不想无限等下去，比如"最多等 200 毫秒，等不到就先去做点别的事情，过会儿再试"，这时候就需要 std::timed_mutex。
它比普通 mutex 多提供了两个方法：

• try_lock_for(时长)：最多尝试等待这么久，如果时间到了还没抢到锁，就直接返回 false，不会继续傻等。
• try_lock_until(时间点)：效果类似，只是描述"等到什么时候"用的是一个绝对时间点。

#include <iostream>
#include <mutex>
#include <thread>
#include <chrono>
std::timed_mutex g_timed_mtx; // 支持超时尝试加锁的互斥量
void long_task() {
    // 先让这个线程长时间占用锁，模拟"锁被别人长期占用"的情况
    std::lock_guard<std::timed_mutex> lock(g_timed_mtx);
    std::cout << "长任务线程拿到了锁，开始长时间占用" << std::endl;
    std::this_thread::sleep_for(std::chrono::seconds(2));
    std::cout << "长任务线程释放锁" << std::endl;
}
void impatient_task() {
    // 最多只愿意等待 500 毫秒
    // try_lock_for 返回 true 表示成功拿到锁，返回 false 表示超时仍未拿到
    if (g_timed_mtx.try_lock_for(std::chrono::milliseconds(500))) {
        std::cout << "没耐心的线程成功拿到了锁" << std::endl;
        g_timed_mtx.unlock(); // 手动加锁的方式必须手动解锁
    } else {
        std::cout << "没耐心的线程等了 500 毫秒仍未拿到锁，选择放弃，去做别的事情" << std::endl;
    }
}
int main() {
    std::thread t1(long_task);
    // 稍微等一下，确保 t1 已经先拿到了锁
    std::this_thread::sleep_for(std::chrono::milliseconds(100));
    std::thread t2(impatient_task);
    t1.join();
    t2.join();
    return 0;
}

关键点说明：

• try_lock_for 和 try_lock_until 返回一个 bool，一定要检查这个返回值，只有返回 true 才代表真正拿到了锁，才能安全地进入临界区并且需要负责后续解锁；如果直接用 std::lock_guard 包装 try_lock_for 的结果会比较麻烦，实际项目中更常见的写法是用 std::unique_lock 配合 std::try_to_lock 标签，这里为了讲清楚原理，先演示最直接的手动 lock/unlock 方式。
  编译运行：

g++ -std=c++17 -pthread timed_mutex_demo.cpp -o timed_mutex_demo
./timed_mutex_demo

6. std::recursive_timed_mutex 与 std::shared_timed_mutex：能力的组合

剩下这两种类型，与其说是全新的概念，不如说是前面几种能力的组合：

• std::recursive_timed_mutex = 递归加锁能力 + 超时加锁能力。也就是说，它既允许同一个线程重复加锁多次（像 recursive_mutex 那样），又提供 try_lock_for/try_lock_until 这种带超时的尝试加锁方式（像 timed_mutex 那样）。
• std::shared_timed_mutex = 共享/独占两种访问模式 + 超时加锁能力。也就是说，它既支持"多个线程同时共享读"、“一个线程独占写”（像 shared_mutex 那样），共享模式和独占模式各自也都提供了带超时的尝试加锁接口（像 timed_mutex 那样）。
  用一小段代码演示 std::recursive_timed_mutex 的组合效果：

#include <iostream>
#include <mutex>
#include <thread>
#include <chrono>
std::recursive_timed_mutex g_rec_timed_mtx;
void recursive_with_timeout(int depth) {
    // 既可以像 recursive_mutex 那样在同一线程内反复加锁
    if (g_rec_timed_mtx.try_lock_for(std::chrono::milliseconds(300))) {
        std::cout << "第 " << depth << " 层成功加锁" << std::endl;
        if (depth < 3) {
            recursive_with_timeout(depth + 1); // 同一线程递归加锁，不会死锁
        }
        g_rec_timed_mtx.unlock(); // 每一层加锁都要对应一次解锁
    } else {
        std::cout << "第 " << depth << " 层等待超时，放弃加锁" << std::endl;
    }
}
int main() {
    std::thread t(recursive_with_timeout, 1);
    t.join();
    return 0;
}

编译运行：

g++ -std=c++17 -pthread recursive_timed_mutex_demo.cpp -o recursive_timed_mutex_demo
./recursive_timed_mutex_demo

因为 shared_timed_mutex 只是在 shared_mutex 的基础上，把第 4 节里的 std::shared_lock/std::unique_lock 换成支持超时版本的加锁尝试（比如 try_lock_shared_for、try_lock_for），思路和上面完全一致，这里不再重复给出完整代码。

7. 该怎么选：一张决策图

8. 小结

六种互斥量类型全部都是围绕"独占还是共享"“是否支持同一线程递归加锁”“是否支持超时尝试"这三个维度自由组合出来的。绝大多数日常场景直接用最基础的 std::mutex 就足够；只有当确实存在"同一线程需要重复加锁”（选 recursive）、“读远多于写”（选 shared）、“不想无限期等待”（选 timed）这类明确需求时，才需要用到对应的变体，也可以按需组合出带多个能力的版本。

超时互斥量：try_lock_for 与 try_lock_until

前面接触过的互斥量在"抢锁"这件事上，行为其实都差不多：

• 调用 lock()：如果锁已经被别人占用，当前线程就一直阻塞等着，直到抢到锁为止，没有时间上限。
• 调用 try_lock()：立刻尝试一次，抢到就返回 true，抢不到立刻返回 false，不会等待。
  lock() 的问题在于：如果这把锁被别的线程占用了很久，调用 lock() 的线程可能要跟着傻等很长时间，什么别的事都干不了。有些场景下我们更希望的效果是：“最多等一段时间，等不到就算了，先去做点别的事情，稍后再来试”。这正是**超时互斥量（Timed Mutex）**要解决的问题。

1. 三种超时互斥量

C++ 标准库提供了三个带超时能力的互斥量类：std::timed_mutex、std::recursive_timed_mutex、std::shared_timed_mutex。它们分别是普通 mutex、recursive_mutex、shared_mutex 的"超时增强版"，除了原来就有的 lock() 和 try_lock()，还额外提供了两个新方法：

• try_lock_for(时长)：尝试加锁，最多等待指定的这段时长。如果在这段时间结束之前成功抢到了锁，返回 true；如果时间到了还没抢到，返回 false，线程不会继续傻等下去。
• try_lock_until(时间点)：效果和 try_lock_for 类似，只是描述"等到什么时候放弃"的方式不同——不是给一段"时长"，而是给一个未来的"绝对时间点"，一直等到这个时间点，或者提前抢到锁，两者谁先发生就按谁算。

2. 两个需要特别注意的细节

关于 try_lock_for，有两个容易被忽略、但很重要的行为规则：
第一，如果传入的时长小于等于零（也就是 timeout_duration.zero() 成立），这个函数的行为会退化成和 try_lock()完全一样：只尝试抢一次，抢不到立刻返回 false，不会有任何等待。
第二，实际阻塞的时间可能比你指定的更长。这一点和前面章节讲过的 sleep_for 是同样的道理：受操作系统调度安排、或者多个线程同时争抢同一把锁造成的资源竞争影响，try_lock_for 保证的是一个"下限"，而不是精确的截止时间，可以理解成：
$$实际等待时长\ge 指定的超时时长$$

3. 完整代码示例：用 try_lock_for 实现"抢不到就转做别的事"

下面这个例子开了 8 个线程，每个线程反复尝试对同一把 std::timed_mutex 加锁：抢到了就把计数器加一并打印；抢不到，就转而用另外一把普通互斥量保护"失败计数"并打印一条失败信息。

#include <algorithm> // 本例虽未直接用到具体算法，但保留和原场景一致的常用头文件
#include <chrono>    // 时间字面量 比如 10ms
#include <iostream>  // 标准输入输出
#include <mutex>     // std::timed_mutex 和 std::mutex 都在这里
#include <thread>    // std::thread
#include <vector>    // 存放多个线程对象
constexpr int NUM_THREADS = 8; // 一共开 8 个线程互相竞争
int counter = 0; // 抢锁成功的次数，由 tm 这把超时互斥量保护
int failed = 0;  // 抢锁失败的次数，由 m 这把普通互斥量保护
int main() {
    using namespace std::chrono_literals; // 之后可以直接写 10ms 这种字面量
    std::timed_mutex tm; // 支持超时尝试的互斥量，用来保护 counter 以及对应的打印
    std::mutex m;        // 普通互斥量，用来保护 failed 以及对应的打印
    // 每个线程要反复执行的任务，用 Lambda 表达式定义，[&] 表示按引用捕获外部变量
    auto worker = [&] {
        for (int i = 0; i < 10; ++i) {
            // 尝试对 tm 加锁，最多愿意等待 10 毫秒
            if (tm.try_lock_for(10ms)) {
                // 走到这里说明成功抢到了 tm 这把锁
                ++counter;
                std::cout << "计数器: " << counter << std::endl;
                // 模拟持有锁期间做一点耗时的工作
                std::this_thread::sleep_for(10ms);
                // 注意：这里一定要释放刚才成功抢到的那把锁 tm，
                // 如果这里误写成 m.unlock()，因为 m 在这个分支里根本没有被加锁过，
                // 对一把没有被当前线程持有的锁调用 unlock() 属于未定义行为，
                // 而且 tm 也永远不会被释放，后续所有线程都再也抢不到它了
                tm.unlock();
            } else {
                // 走到这里说明 10 毫秒之内没能抢到 tm，选择放弃，转而处理"失败计数"
                // 这里用的是完全独立的另一把锁 m，因为 failed 变量和 counter 是两份不同的数据，
                // 分别用两把不相关的锁保护，可以避免它们互相之间没必要的等待
                m.lock();
                ++failed;
                std::cout << "线程 " << std::this_thread::get_id()
                          << " 加锁失败" << std::endl;
                m.unlock();
            }
            // 每一轮循环结束后，不管成功还是失败，都统一睡眠 12 毫秒，
            // 让各个线程的抢锁节奏错开一些，方便观察效果
            std::this_thread::sleep_for(12ms);
        }
    };
    std::vector<std::thread> threads;
    // 创建 8 个线程，全部执行同一份 worker 逻辑，彼此激烈地争抢 tm 这把锁
    for (int i = 0; i < NUM_THREADS; ++i) {
        threads.emplace_back(worker);
    }
    for (auto& t : threads) {
        t.join();
    }
    std::cout << "最终计数器: " << counter << std::endl;
    std::cout << "总失败次数: " << failed << std::endl;
    return 0;
}

关键点说明：

• 代码里同时用了两把完全独立的锁：tm（timed_mutex，保护 counter 和对应的打印）和 m（普通 mutex，保护 failed 和对应的打印）。这样设计的原因是 counter 和 failed 是两份互不相干的数据，各自只需要保护好自己就行，没必要用同一把锁把它们捆在一起，捆在一起反而会让原本可以独立进行的操作互相多等待。
• 抢锁成功的分支里，一定要记得对成功抢到的那把锁（也就是 tm）调用 unlock()，如果误调用了 m.unlock()，会产生两个问题：一是 m 根本没被这个分支加锁过，对它调用 unlock() 是未定义行为；二是真正被抢到的 tm 永远没有被释放，后续所有线程的 try_lock_for 都会持续失败。这是使用超时互斥量时最容易踩的坑——一定要保证"谁加的锁，就由谁负责解锁"。
• 因为 8 个线程在激烈竞争同一把 tm，实际运行时 counter 的最终值通常会明显小于 8 × 10 = 80（8 个线程各循环 10 次），因为每一轮总有一部分线程会因为 10 毫秒内抢不到锁而转入失败分支；具体数值每次运行都可能不一样，这是正常现象。
  编译运行方式：

g++ -std=c++17 -pthread timed_mutex_race_demo.cpp -o timed_mutex_race_demo
./timed_mutex_race_demo

4. 用时序图理解单个线程一轮循环的分支走向

5. try_lock_for 与 try_lock_until 对比

---

特性	try_lock_for	try_lock_until
参数类型	一段相对时长 duration	一个绝对时间点 time_point
语义	最多再等这么久	最多等到这个时间点为止
传入零或负时长 负值时长的处理	时长为零时退化成 try_lock 效果	时间点已经是过去时刻时同样退化成try_lock效果
实际阻塞时间是否可能超出预期	可能 受调度和资源竞争影响	可能 同样受调度和资源竞争影响
适合的场景	已知"最多愿意等多久" 比如固定的重试间隔	需要在某个具体截止时间点之前完成加锁 比如统一的超时截止点

---

6. 小结

try_lock_for 和 try_lock_until 给了我们在"无限等待"和"完全不等待"之间的一个折中选项：设定一个可接受的等待上限，超过这个上限就主动放弃、转而处理别的逻辑，而不是让线程干等着白白浪费时间。使用时要记住两个要点：一是这个等待上限只是下限保证，实际阻塞时间可能因为调度或资源竞争而更长；二是一定要在正确的那把锁上调用 unlock()，成功抢到哪把锁，就必须由这把锁自己负责释放，绝不能张冠李戴，否则会造成锁永远无法释放或者对未加锁的互斥量做出未定义行为这类隐蔽的严重问题。

通用锁管理：四个让互斥量更好用的包装类

前面已经认识了标准库提供的各种互斥量类型。这一节要讲的不是新的互斥量，而是怎样更安全、更方便地使用它们——通过几个"包装类"来自动管理加锁和解锁这件事，而不是每次都手动调用 lock() 和 unlock()。

1. 为什么不能只靠手动 lock/unlock

如果每次都手写 mtx.lock(); ...; mtx.unlock();，会有一个很大的隐患：只要中间那段代码在执行过程中提前 return，或者抛出了异常，unlock() 这一行就会被跳过，导致这把锁永远不会被释放，其他所有还在等这把锁的线程都会被永久卡住。
标准库提供的这几个包装类都遵循同一个思路——RAII（资源获取即初始化）：构造这个包装对象的时候自动加锁，这个对象离开作用域被销毁的时候自动解锁，不管是正常执行完走出作用域，还是中途异常导致栈展开，析构函数都一定会被调用，所以锁一定能被正确释放，不用担心漏写 unlock()。

2. 四个锁管理类总览

---

管理类	支持的互斥量类型	同时能管理几把锁
std::lock_guard	所有互斥量类型都可以	1把
std::scoped_lock	所有互斥量类型都可以	0把或任意多把
std::unique_lock	所有互斥量类型都可以	1把
std::shared_lock	仅限std::shared_mutex和std::shared_timed_mutex	1把

---

用一棵树梳理一下它们的定位关系：

锁管理包装类
├── 只管一把锁 只负责独占模式加锁
│   ├── std::lock_guard 最简单 功能最少 开销最低
│   └── std::unique_lock 功能最全 支持延迟加锁 手动解锁 转移所有权
├── 能同时管理多把锁
│   └── std::scoped_lock 一次性锁住零把或多把互斥量 内部自带死锁避免机制
└── 只管一把锁 但负责共享模式加锁
    └── std::shared_lock 专门配合shared_mutex系列使用

3. std::lock_guard：最简单、最轻量的选择

std::lock_guard 是这几个包装类里功能最少、也最简单的一个：构造时加锁，析构时解锁，仅此而已，不提供手动提前解锁、重新加锁之类的额外功能。正因为功能简单，它的开销也是这几个类里最小的，如果只是想"进入一段代码前加锁，出去自动解锁"，lock_guard 通常是首选。

#include <iostream>
#include <mutex>
#include <thread>
std::mutex g_mtx;
int g_value = 0;
void increment() {
    // 构造 lock_guard 的同时立刻对 g_mtx 调用 lock()
    std::lock_guard<std::mutex> guard(g_mtx);
    ++g_value;
    // 这里没有任何手动 unlock 的代码，
    // 当 guard 在函数结束时被销毁，它的析构函数会自动调用 g_mtx.unlock()
} // guard 在这里被销毁 自动解锁
int main() {
    std::thread t1(increment);
    std::thread t2(increment);
    t1.join();
    t2.join();
    std::cout << "结果: " << g_value << std::endl;
    return 0;
}

编译运行：

g++ -std=c++17 -pthread lock_guard_demo.cpp -o lock_guard_demo
./lock_guard_demo

4. std::scoped_lock：一次性安全地锁住多把锁

如果一段代码需要同时持有两把或更多把锁，直接对它们分别用 lock_guard 依次加锁，是有风险的。举个经典的例子：账户转账场景，从账户 A 转钱到账户 B 需要同时锁住 A 和 B 两把互斥量；如果线程 1 按"先锁 A 再锁 B"的顺序加锁，而线程 2（正好在做反方向的转账）按"先锁 B 再锁 A"的顺序加锁，两个线程就可能同时各自锁住一把、又在等待对方手里的另一把，谁都不肯放手，形成死锁。
用一张时序图看看这种死锁是怎么发生的：

std::scoped_lock（C++17 引入）就是为了从根本上避免这种情况：它可以在一次构造调用里同时锁住多把互斥量，内部使用了死锁避免算法（本质上是保证所有线程都用一种统一、不会产生循环等待的顺序去尝试获取这些锁），不管调用者传入互斥量的先后顺序是怎样的，都不会因为"加锁顺序不同"而死锁。它还支持传入零个互斥量，这种情况下它什么也不做，通常用于泛型代码里统一写法、避免特殊情况判断。

#include <iostream>
#include <mutex>
#include <thread>
std::mutex g_mutex_a;
std::mutex g_mutex_b;
// 模拟从一个账户转账到另一个账户，需要同时锁住两个账户对应的互斥量
void transfer(std::mutex& from, std::mutex& to, const std::string& label) {
    // scoped_lock 在构造时一次性把 from 和 to 两把锁都锁住，
    // 不管传入顺序如何，内部都会用统一的策略去获取，避免和别的线程发生循环等待
    std::scoped_lock lock(from, to);
    std::cout << label << " 同时持有了两把锁，正在处理转账逻辑" << std::endl;
    // 这里可以安全地同时操作两个账户各自的数据
} // lock 在这里被销毁，两把锁按安全的顺序依次自动释放
int main() {
    // 线程1模拟"从A到B"的转账，线程2模拟"从B到A"的转账，加锁参数顺序故意相反
    std::thread t1(transfer, std::ref(g_mutex_a), std::ref(g_mutex_b), "线程1 A转B");
    std::thread t2(transfer, std::ref(g_mutex_b), std::ref(g_mutex_a), "线程2 B转A");
    t1.join();
    t2.join();
    std::cout << "两笔转账都安全完成，没有发生死锁" << std::endl;
    return 0;
}

关键点说明：

• 即使 t1 是"先 A 后 B"、t2 是"先 B 后 A"这种刻意制造死锁风险的加锁顺序，scoped_lock 依然能保证两个线程都能顺利往下走，不会卡死，这正是它相比"分别用两个 lock_guard"最大的优势。
  编译运行：

g++ -std=c++17 -pthread scoped_lock_demo.cpp -o scoped_lock_demo
./scoped_lock_demo

5. std::unique_lock：功能最全面的独占锁包装

std::unique_lock 同样一次只管理一把锁，但相比 lock_guard，它提供了灵活得多的功能：

• 可以先构造对象但不立刻加锁（配合 std::defer_lock 标签），之后手动决定什么时候加锁。
• 可以手动调用 lock()、unlock()、try_lock()，中途想提前解锁、之后再重新加锁都可以。
• 可以被移动（转移锁的所有权到另一个 unique_lock 对象），这一点 lock_guard 是做不到的。
• 是配合 std::condition_variable 使用时唯一被接受的锁类型（之前讲条件变量时用到的 unique_lock 正是因为 wait() 需要中途释放锁和重新加锁的能力，而这正是 lock_guard 不具备、unique_lock 具备的能力）。

#include <iostream>
#include <mutex>
#include <thread>
std::mutex g_mtx;
int g_value = 0;
void flexible_task() {
    // 用 std::defer_lock 标签构造，表示"先不要立刻加锁"
    std::unique_lock<std::mutex> lock(g_mtx, std::defer_lock);
    std::cout << "此时还没有加锁，可以先做一些不需要保护的准备工作" << std::endl;
    // 需要的时候再手动加锁
    lock.lock();
    ++g_value;
    std::cout << "已加锁并修改了共享数据" << std::endl;
    // 可以在离开作用域之前就提前手动解锁，让别的线程尽快有机会进入
    lock.unlock();
    std::cout << "已经提前手动解锁，这里可以继续做其他不需要保护的事情" << std::endl;
} // 因为上面已经手动 unlock 过，这里析构时不会重复解锁
int main() {
    std::thread t(flexible_task);
    t.join();
    std::cout << "结果: " << g_value << std::endl;
    return 0;
}

关键点说明：

• unique_lock 内部会记录自己当前"是否持有锁"的状态，即使中途手动 unlock() 过，析构时也不会重复解锁，不会出现"多解锁一次"这种错误。
• 相比 lock_guard，unique_lock 因为要维护这些额外状态，会有一点点性能开销，所以如果不需要这些灵活功能，lock_guard 仍然是更轻量的首选；只有确实需要"延迟加锁"“手动控制加解锁时机”"配合条件变量"这类需求时，才应该选择 unique_lock。
  编译运行：

g++ -std=c++17 -pthread unique_lock_demo.cpp -o unique_lock_demo
./unique_lock_demo

6. std::shared_lock：专门配合读写锁的共享模式包装

std::shared_lock 的用法和 unique_lock 很像，但它只能配合 std::shared_mutex 或 std::shared_timed_mutex 使用，而且它加的是共享模式的锁，也就是允许多个线程同时持有。前面讲 shared_mutex 的时候已经用到过它了，这里再单独强调一下它的定位：

#include <iostream>
#include <shared_mutex>
#include <thread>
std::shared_mutex g_shared_mtx;
int g_data = 42;
void read_task(int id) {
    // shared_lock 加的是共享模式的锁，多个线程可以同时构造并持有它
    std::shared_lock<std::shared_mutex> lock(g_shared_mtx);
    std::cout << "读线程 " << id << " 读到的数据: " << g_data << std::endl;
} // 离开作用域自动释放共享锁
int main() {
    std::thread t1(read_task, 1);
    std::thread t2(read_task, 2);
    t1.join();
    t2.join();
    return 0;
}

编译运行：

g++ -std=c++17 -pthread shared_lock_demo.cpp -o shared_lock_demo
./shared_lock_demo

7. 该用哪个：一张决策图

8. 小结

这四个锁管理类都基于同一个核心思想——RAII，靠对象的生命周期自动保证锁一定会被释放，避免手动 lock/unlock 时因为异常或提前返回而漏掉解锁。日常最简单的场景优先用 lock_guard；需要同时锁住多把互斥量、又想避免因加锁顺序不同而死锁时，用 scoped_lock；需要更灵活地控制加解锁时机、或者要配合条件变量时，用 unique_lock；如果互斥量本身是支持读写分离的 shared_mutex，并且只是想以共享模式读取，就用 shared_lock。

std::lock_guard 详解

前面学习 std::mutex 的时候，用的都是手动调用 lock() 和 unlock() 的写法。这种写法有一个隐患：如果在 lock() 和 unlock() 之间的代码抛出了异常，程序会直接从那个位置跳出去，unlock() 这一行代码根本没有机会被执行到，锁就这样被"永远地"留在了加锁状态。这一节要介绍的 std::lock_guard，正是用来彻底解决这个问题的。

1. 手动管理锁遇到异常时会发生什么

先想清楚问题出在哪。假设代码是这样写的：

mtx.lock();
counter++;
function_throws(); // 这里抛出了一个异常
mtx.unlock();       // 因为上一行抛了异常，这一行永远不会被执行到

一旦 function_throws() 抛出异常，程序的控制流会立刻"跳走"，去寻找能够处理这个异常的 catch 块，中间这一行 mtx.unlock(); 就这样被跳过了，锁 mtx 会一直停留在"被锁住"的状态，再也没有人来释放它。如果后面还有别的线程想申请这把锁，就会永远阻塞在那里等待——这实际上制造了一种"事故性"的死锁，而且非常隐蔽，因为代码表面上看起来 lock() 和 unlock() 是成对写的，很容易被忽略掉"中间可能会抛异常"这个风险点。

1.1 用图对比"有没有异常保护"的两种结局

从这张图能看出，问题的关键在于：手动写的 unlock() 是一行"普通代码"，只有正常执行到那一行才会生效；而 RAII 工具的释放动作，是绑定在对象的析构函数上的，只要对象的生命周期结束（不管是正常执行完，还是因为异常导致栈展开），析构函数都一定会被调用。

2. std::lock_guard 是什么

std::lock_guard 是一个遵循 RAII（资源获取即初始化，Resource Acquisition Is Initialization） 原则设计的类，它的作用非常单纯：让互斥锁的使用变得更简单、更安全，保证互斥锁一定会在 lock_guard 对象被析构的时候被释放。这个特性在处理异常的场景下特别有价值，因为不管代码是正常走完，还是中途因为异常被迫提前退出，只要 lock_guard 对象的生命周期结束了（离开了它所在的作用域），它的析构函数就一定会被调用，锁也就一定会被释放。
lock_guard 的用法非常简单：

std::lock_guard<std::mutex> lock(mtx);

这一行代码做了两件事：

1. 构造 lock_guard 对象的同时，自动调用 mtx.lock()，把锁申请下来。
2. 把这把锁"托管"给这个 lock 对象，之后完全不需要再手动调用 mtx.unlock()——只要 lock 这个对象的生命周期结束（比如离开当前作用域），它的析构函数就会自动帮你调用 mtx.unlock()。

3. 完整代码示例：lock_guard 如何简化异常处理

下面这段代码同时展示了两种线程的写法：一个线程用最原始的手动 lock()/unlock()；另一个线程用 lock_guard，并且函数内部会主动抛出一个异常，用来验证 lock_guard 在异常场景下依然能正确释放锁。

#include <iostream>   // 标准输入输出
#include <mutex>      // std::mutex、std::lock_guard
#include <thread>     // std::thread
#include <stdexcept>  // std::runtime_error
#include <system_error> // std::system_error
std::mutex mtx;           // 保护counter的互斥锁
uint32_t counter{};        // 被两个线程共享的计数器，初始化为0
// 一个简单的工具函数：调用它必定会抛出一个运行时异常
void function_throws() {
    throw std::runtime_error("Error");
}
int main() {
    // 线程1要执行的任务：手动lock/unlock（没有异常，正常累加）
    auto worker = [] {
        for (int i = 0; i < 1000000; ++i) {
            mtx.lock();
            counter++;
            mtx.unlock();
        }
    };
    // 线程2要执行的任务：用lock_guard，并且每次循环都会触发一次异常
    auto worker_exceptions = [] {
        for (int i = 0; i < 1000000; ++i) {
            try {
                // 构造lock_guard的同时，自动完成 mtx.lock()
                std::lock_guard<std::mutex> lock(mtx);
                counter++;           // 临界区内的操作
                function_throws();   // 这一行必定抛出 std::runtime_error
                // 注意：下面这行代码，以及本次循环剩余的部分，
                // 由于上一行抛出了异常，永远不会被执行到
            } catch (std::system_error& e) {
                // 这个分支专门用来捕获 std::system_error 类型的异常
                // 但function_throws抛出的是std::runtime_error，并不是system_error
                // 所以这个分支实际上不会被触发，这里写出来只是展示"可以有针对性的catch"
                std::cout << e.what() << std::endl;
                return;
            } catch (...) {
                // "..."表示捕获任意类型的异常，因为上面的分支没接住，
                // 真正抛出的 std::runtime_error 会在这里被捕获
                // 不管走到这个catch块的哪一步，try块里构造的lock对象
                // 都已经在"栈展开"的过程中被销毁，mtx也已经被自动释放了
                return; // 直接结束这个lambda，不再继续循环
            }
        }
    };
    std::thread t1(worker_exceptions);
    std::thread t2(worker);
    t1.join();
    t2.join();
    std::cout << "Final counter value: " << counter << std::endl;
    return 0;
}

关键点解析：

• worker 这个线程用的是最朴素的手动加锁方式，因为它内部没有任何会抛异常的代码，所以 lock()/unlock() 能够按预期正常配对执行，不会有问题。
• worker_exceptions 这个线程每次循环都会执行 function_throws()，这个函数必定抛出一个 std::runtime_error 异常。正因为如此，这个循环实际上只会真正执行一次（counter++ 只加了一次），第一次调用 function_throws() 抛出异常之后，就会被下面的 catch (...) 接住，然后 return 直接退出整个 lambda，不会继续跑满一百万次循环。
• 最关键的一点在于：std::lock_guard<std::mutex> lock(mtx); 这个 lock 对象是在 try 块内部、function_throws() 之前定义的局部变量。当 function_throws() 抛出异常后，程序会进行栈展开（stack unwinding）——也就是从异常抛出的位置开始，沿着函数调用栈往外"退"，依次销毁沿途所有已经构造完成的局部对象。lock 正是这样一个局部对象，在栈展开的过程中，它的析构函数会被自动调用，从而自动执行了相当于 mtx.unlock() 的操作。这个释放过程完全不需要我们写任何额外的代码去处理，lock_guard 已经把这件事安排妥当了。
• 这里还有一个值得注意的细节：代码里第一个 catch (std::system_error& e) 分支实际上永远不会被执行到，因为 function_throws() 抛出的是 std::runtime_error，而不是 std::system_error，两者是不同的异常类型，真正起作用、接住这个异常的是后面的 catch (...)。这提醒我们写异常处理代码时，一定要清楚自己捕获的异常类型是否真的和抛出的类型匹配。

3.1 lock_guard 生命周期与栈展开的时序图

从图上可以看到，lock 对象的销毁（也就是锁的自动释放）发生在异常真正被 catch 块捕获处理之前，这正是 RAII 机制的巧妙之处：不管异常最终会被谁接住、接住之后要做什么，沿途所有局部对象的清理工作都会先一步、自动地完成。

4. std::lock_guard 的另一个构造函数：std::adopt_lock

除了最常见的"构造时自动加锁"这种用法，std::lock_guard 还提供了另外一个构造函数，接收一个 std::adopt_lock_t 类型的参数（实际使用时直接传入 std::adopt_lock 这个预定义的标记对象即可）。
这个构造函数的作用是：包装一把已经被手动加锁过的互斥锁，也就是说，锁已经在别的地方被 lock() 过了，我们只是想借助 lock_guard 的析构函数，帮忙在合适的时机自动完成 unlock()，而不需要它在构造的时候再重新加一次锁。

#include <iostream>
#include <mutex>
#include <thread>
std::mutex mtx;
void demo_adopt_lock() {
    mtx.lock(); // 先手动加锁
    // 这里用 std::adopt_lock 标记告诉 lock_guard：
    // "这把锁已经被加过了，你不需要再调用一次lock()，
    //  只需要在你(lock_guard对象)被销毁的时候帮我调用unlock()就行"
    std::lock_guard<std::mutex> lock(mtx, std::adopt_lock);
    std::cout << "在临界区内处理数据..." << std::endl;
    // 函数结束时，lock对象被销毁，自动调用mtx.unlock()
    // 不需要我们再手动写一行 mtx.unlock();
}
int main() {
    std::thread t(demo_adopt_lock);
    t.join();
    return 0;
}

关键点解析：

• 普通用法 std::lock_guard<std::mutex> lock(mtx); 里，lock_guard 在构造时主动调用了 mtx.lock()；而这里用 std::adopt_lock 标记的写法，lock_guard 在构造时不会再调用 lock()，它只是"接手"这把已经处于加锁状态的锁，负责在自己析构的时候调用 unlock()。
• 这种用法常见于一些锁已经在别处（比如通过 std::lock() 这种可以同时安全锁住多把互斥量的函数）被加上了，但又希望后续的释放操作能借助 RAII 机制自动完成、不容易漏写的场景。

4.1 两种构造方式对比

---

构造方式	写法	构造时是否调用lock()	析构时是否调用unlock()
默认构造	std::lock_guard<std::mutex> lock(mtx);	会	会
adopt_lock构造	std::lock_guard<std::mutex> lock(mtx, std::adopt_lock);	不会(假设已经被lock过)	会

---

小结

• 手动调用 lock()/unlock() 时，一旦中间的代码抛出异常，unlock() 就会被跳过，导致锁永久处于占用状态，这是一个很容易被忽略的隐患。
• std::lock_guard 遵循 RAII 原则，构造时自动加锁，析构时自动解锁，不管是正常执行完毕，还是因为异常导致栈展开提前退出作用域，锁都一定会被正确释放，从根本上解决了"忘记unlock"的问题。
• 使用 lock_guard 之后，代码里完全不需要再手写 unlock()，只要把需要保护的代码放在 lock_guard 对象存活的作用域内即可，写法更简洁，也更不容易出错。
• 除了最常用的"构造时自动加锁"这种方式，std::lock_guard 还提供了配合 std::adopt_lock 使用的构造函数，用于包装一把已经被提前手动加锁的互斥量，只借助它的析构函数来自动完成释放。

std::unique_lock：比 lock_guard 更灵活的锁管理工具

一、先回顾一下 std::lock_guard 的局限

std::lock_guard 是一个非常简单的 std::mutex 包装器：构造的时候自动调用 lock() 申请锁（如果锁被别人占着，当前线程就在这里阻塞等待），析构的时候自动调用 unlock() 释放锁。这种"构造即加锁、析构即解锁"的写法，靠的是 C++ 的 RAII 机制（资源获取即初始化），好处是不容易忘记解锁，即使中间抛了异常，锁也能在栈展开的过程中被自动释放。
但 lock_guard 的行为非常"死板"，只支持一种模式：构造时无条件调用 lock()（或者假设锁已经被拿到了），析构时无条件调用 unlock()。如果实际需求稍微复杂一点，比如：

• 不想在构造的时候就立刻申请锁，而是想晚一点、条件满足了再手动加锁；
• 想用 try_lock() 去"试一下"能不能拿到锁，拿不到就干别的事，而不是傻等；
• 拿到锁之后，中途想主动解锁一下，做点不需要保护的事情，然后再重新加锁；
  这些需求 lock_guard 统统做不到，因为它连 lock()、unlock() 这样的成员函数都没有对外暴露。这时候就需要用到功能更全面的 std::unique_lock。
  用一张图对比两者的能力范围：

二、std::unique_lock 的构造方式：三种"标签"参数

std::unique_lock 的构造函数支持传入第二个参数——一个特殊的标签类型，用来告诉它"对这把 mutex 具体要做什么处理"。一共有三种可选的标签：

• std::defer_lock：表示"暂时不要在构造的时候加锁"。构造函数只是把这把 mutex "记"下来，并不会立刻调用 lock()。既然构造时没有加锁，如果后续这把锁真的从来没被这个 unique_lock 对象获取过，那么析构的时候也就不会去调用 unlock()（没锁过，自然也不用解锁）。
• std::adopt_lock：表示"这把 mutex 我这个线程其实已经提前拿到手了，你不用再帮我 lock() 一次"。unique_lock 只是接管这把已经被持有的锁的所有权，负责在析构的时候帮忙调用 unlock()。这个标签 std::lock_guard 也支持。
• std::try_to_lock：表示"帮我用 try_lock() 的方式尝试获取这把锁，不要阻塞等待"。如果暂时拿不到，构造函数不会卡住，而是让这个 unique_lock 对象记住"这次没拿到锁"这个状态，后续可以通过 owns_lock() 之类的方法去判断到底有没有真的拿到。
  如果构造 unique_lock 的时候只传一个 mutex、不传标签，它的行为就和 lock_guard 一模一样：阻塞式地调用 lock()，构造成功就意味着锁已经到手，析构的时候自动 unlock()。
  用表格总结一下这几种用法的区别：

---

构造方式	构造时是否阻塞加锁	适用场景
unique_lock(mtx)（不传标签）	是，行为等同于 lock_guard	只是想要 RAII 风格的自动加解锁，没有特殊需求
unique_lock(mtx, std::defer_lock)	否，暂不加锁	需要先构造对象，之后再根据条件决定何时手动 lock()
unique_lock(mtx, std::adopt_lock)	否，假设锁已经在手上	锁已经被显式 lock() 过，只是想让 unique_lock 接管、负责后续释放
unique_lock(mtx, std::try_to_lock)	尝试获取但不阻塞	想"试一下"能不能拿到锁，拿不到就走别的分支，而不是傻等

---

三、代码逐一演示这三种标签

3.1 std::defer_lock：延迟加锁，需要时再手动 lock()

#include <iostream>
#include <mutex>
#include <thread>
std::mutex mtx;
void deferredExample() {
    // 关键点：这里构造 unique_lock 时传入了 std::defer_lock，
    // 意味着这一行代码执行完，锁并没有被真正获取，
    // ul 这个对象只是"记住了"要管理的是哪把 mutex。
    std::unique_lock<std::mutex> ul(mtx, std::defer_lock);
    std::cout << "还没加锁，可以先做一些不需要保护的准备工作……" << std::endl;
    // 一些不涉及共享资源的准备逻辑可以放在这里，不需要占着锁
    // 关键点：真正需要保护共享资源的时候，手动调用 lock()，
    // 这一步才会真正阻塞等待并获取锁。
    ul.lock();
    std::cout << "现在才真正加锁，开始操作共享资源" << std::endl;
    // ul 离开作用域时析构，因为这次确实通过 lock() 拿到了锁，
    // 析构函数会自动调用 unlock() 释放它。
}
int main() {
    std::thread t1(deferredExample);
    std::thread t2(deferredExample);
    t1.join();
    t2.join();
    return 0;
}

讲解：std::defer_lock 解决的是"我现在还不想加锁，但希望用 unique_lock 的 RAII 特性帮我管理这把锁的生命周期"这个需求。构造完成后，ul 对象处于"没有持有锁"的状态，直到显式调用 ul.lock()，才真正去申请这把锁；一旦申请成功，之后 ul 析构时就会自动释放。

3.2 std::adopt_lock：接管一把已经拿到手的锁

#include <iostream>
#include <mutex>
#include <thread>
std::mutex mtx;
void adoptExample() {
    mtx.lock(); // 先手动加锁，此时这把锁已经被当前线程持有
    std::cout << "已经手动获取了锁，接下来交给 unique_lock 管理释放" << std::endl;
    // 关键点：std::adopt_lock 告诉 unique_lock"这把锁我已经拿到手了，
    // 你不需要再帮我 lock() 一次（如果再 lock() 一次，同一线程对普通 mutex
    // 重复加锁会导致死锁，参考前面讲的 recursive_mutex 那一节）。
    // unique_lock 构造完成后，接管的只是"负责在析构时释放这把锁"这个责任。
    std::unique_lock<std::mutex> ul(mtx, std::adopt_lock);
    std::cout << "继续操作共享资源……" << std::endl;
    // ul 离开作用域时自动调用 unlock()，不需要再手动写 mtx.unlock()
}
int main() {
    std::thread t1(adoptExample);
    std::thread t2(adoptExample);
    t1.join();
    t2.join();
    return 0;
}

讲解：std::adopt_lock 适合这种场景——锁已经通过别的途径（比如显式调用 mtx.lock()）被拿到手了，只是希望后续交给 unique_lock 来负责"记得释放"这件事，避免自己手动写 unlock() 时忘记、或者中间抛异常导致锁没被释放。这里要特别小心：必须确保这把锁真的已经被当前线程持有，如果锁根本没被拿到就用 adopt_lock 去构造，unique_lock 会误以为自己"接管"了一把锁，析构时去调用 unlock() 释放一把自己根本没持有的锁，这是未定义行为。

3.3 std::try_to_lock：试着拿一下，拿不到就不等

#include <iostream>
#include <mutex>
#include <thread>
#include <chrono>
std::mutex mtx;
void tryLockExample(const std::string& name) {
    // 关键点：std::try_to_lock 让构造函数用非阻塞的方式尝试获取锁，
    // 不管成不成功，这一行代码都会立刻返回，不会卡住当前线程。
    std::unique_lock<std::mutex> ul(mtx, std::try_to_lock);
    // 关键点：owns_lock() 用来判断这次尝试到底有没有真的拿到锁，
    // 只有返回 true，才说明当前对象确实持有这把锁，可以安全操作共享资源。
    if (ul.owns_lock()) {
        std::cout << name << " 成功获取锁，开始工作" << std::endl;
        std::this_thread::sleep_for(std::chrono::milliseconds(200)); // 模拟占用一段时间
    } else {
        std::cout << name << " 没有抢到锁，先去做别的事情" << std::endl;
    }
    // 如果 ul 确实拿到了锁，析构时会自动释放；
    // 如果没拿到，析构时什么都不做，不会误调用 unlock()
}
int main() {
    std::thread t1(tryLockExample, "线程1");
    std::thread t2(tryLockExample, "线程2");
    t1.join();
    t2.join();
    return 0;
}

讲解：std::try_to_lock 适合那种"能拿到锁就干活，拿不到也不想傻等"的场景。这里必须配合 owns_lock() 一起使用，来确认这次构造到底有没有真正拿到锁——因为构造函数本身不会因为没拿到锁而抛异常或者卡住，它只是老老实实地把"有没有拿到"这个结果记在对象内部，交给调用者自己去判断。

四、不传标签：等同于 lock_guard 的默认行为

如果只把 mutex 传给 unique_lock 的构造函数，不附带任何标签：

std::unique_lock<std::mutex> ul(mtx); // 阻塞式加锁，跟 lock_guard 效果一样

这种写法的行为跟 std::lock_guard<std::mutex> lg(mtx); 完全一致：构造函数会阻塞式地调用 lock()，直到成功获取锁为止；析构的时候自动调用 unlock()。区别只是 unique_lock 对象本身会稍微"重"一点点（因为它内部要多维护一些状态信息，比如自己到底有没有持有锁），如果只需要这种最简单的用法，通常优先选 lock_guard，只有在确实需要后面讲的这些灵活功能时，才升级到 unique_lock。

五、unique_lock 独有的能力：手动 lock() / unlock()

std::lock_guard 对外没有暴露任何可以调用的成员函数，一旦构造完成，你唯一能做的事就是等它自然析构。而 std::unique_lock 允许在对象存活期间手动调用 lock() 和 unlock()，中途想解锁一下、稍后再重新加锁，都是允许的：

#include <iostream>
#include <mutex>
#include <thread>
std::mutex mtx;
int shared_data = 0;
void flexibleLocking() {
    std::unique_lock<std::mutex> ul(mtx); // 先正常加锁
    shared_data++; // 操作受保护的共享数据
    ul.unlock(); // 关键点：中途主动解锁，让出机会给其他等待这把锁的线程
    std::cout << "锁已经临时释放，这里可以做一些不需要保护的耗时操作……" << std::endl;
    ul.lock(); // 关键点：需要再次操作共享数据时，重新加锁
    shared_data++; // 再次安全地操作共享数据
    // ul 离开作用域时，因为此刻它确实持有锁（刚刚手动 lock() 过），
    // 析构时会自动调用 unlock() 释放
}
int main() {
    std::thread t1(flexibleLocking);
    std::thread t2(flexibleLocking);
    t1.join();
    t2.join();
    std::cout << "最终 shared_data = " << shared_data << std::endl;
    return 0;
}

讲解：这个例子里，ul.unlock() 主动把锁释放掉，让出这段"不需要保护的耗时操作"的时间窗口，给其他等待这把锁的线程一个机会；等到又要操作共享数据 shared_data 的时候，再调用 ul.lock() 重新申请。这种"锁的持有时间尽量缩短、不需要保护的部分不占着锁"的做法，是编写高并发代码时常见的优化思路，而这正是 lock_guard 做不到、unique_lock 能轻松实现的能力。
用时序图梳理一下这种"中途释放、稍后重新加锁"的过程：

六、小结

std::unique_lock 可以看作是 std::lock_guard 的"加强版"：默认不传标签时，两者行为完全一样；但 unique_lock 额外支持 std::defer_lock（延迟到需要时才手动加锁）、std::adopt_lock（接管一把已经拿到手的锁，只负责后续释放）、std::try_to_lock（非阻塞地试一次能不能拿到锁）这三种更灵活的构造方式，并且允许在对象存活期间随时手动调用 lock()、unlock()，实现"中途释放、按需重新加锁"这类更精细的控制。如果只是最普通的"进入这段代码就加锁，离开就解锁"，lock_guard 已经足够、开销也更小；一旦涉及到延迟加锁、非阻塞尝试或者中途需要临时放开锁这几类需求，就应该换成 unique_lock。

std::scoped_lock 详解

前面学过的 std::lock_guard 有一个明显的局限：它一次只能包装一把互斥锁。如果一个任务需要同时用到两把、三把锁，用 lock_guard 就得写好几行代码分别包装每一把锁，而且一旦不同线程给这些 lock_guard 加锁的顺序不一致，前面讲过的死锁问题依然会发生。std::scoped_lock 正是为了解决"同时安全地管理多把锁"这个问题而设计的。

1. std::scoped_lock 是什么

std::scoped_lock 和 std::lock_guard、std::unique_lock（后面章节会详细介绍）一样，都是 std::mutex 的 RAII 包装器——也就是说，只要锁被成功获取，它就一定会在 scoped_lock 对象被析构的时候自动释放，不需要手动调用 unlock()。
它和 std::lock_guard 最大的区别在于：
$$\text{std::lock\_guard 只能包装 1 把互斥锁}$$
$$\text{std::scoped\_lock 可以包装 0 把或者多把互斥锁}$$
更关键的是，当我们把多把锁一起传给 std::scoped_lock 的构造函数时，它在获取这些锁的过程中，内部使用了一套能够有效避免死锁的算法，这正是本节要重点搞清楚的地方。

2. 一行代码同时安全地拿到两把锁

先看最基础的用法：

std::mutex mtx1;
std::mutex mtx2;
// 一次性、安全地获取两把锁，不会产生死锁
std::scoped_lock lock(mtx1, mtx2);

这一行 std::scoped_lock lock(mtx1, mtx2); 做到的效果，等价于下面这三行代码组合在一起做的事情：

std::lock(mtx1, mtx2); // 用一种能避免死锁的方式，同时锁住mtx1和mtx2
std::lock_guard<std::mutex> lock1(mtx1, std::adopt_lock); // 接手mtx1，负责后续自动unlock
std::lock_guard<std::mutex> lock2(mtx2, std::adopt_lock); // 接手mtx2，负责后续自动unlock

• 第一行 std::lock(mtx1, mtx2); 是标准库提供的一个全局函数，它会用一种特殊的算法同时尝试锁住传入的所有互斥量，并且保证不会因为多把锁而产生死锁（下面第3节会具体解释这套算法的原理）。
• 后面两行用到的正是前一节学过的 std::adopt_lock 用法：因为 mtx1 和 mtx2 已经在第一行被 std::lock 锁住了，这里只是想借助 lock_guard 的析构函数，让这两把锁能在离开作用域时被自动释放，而不需要重新加锁。
  std::scoped_lock 相当于把这三行代码的效果打包封装成了一行，用起来更简洁，也不容易漏写。

3. 为什么 std::scoped_lock 能避免死锁

回顾前面讲过的死锁场景：线程1 按"先锁A、再锁B"的顺序申请，线程2 按"先锁B、再锁A"的顺序申请，两者顺序不一致，就可能各自拿到一把、互相等待对方，形成死锁。
std::scoped_lock（以及它内部使用的 std::lock）从设计上就是为了解决这个问题：不管调用者传入锁的顺序是什么，它内部都会用一套统一的策略去尝试获取这些锁，从而保证不会出现"你等我、我等你"的循环等待。
这套策略大致的思路是：先尝试锁住第一把锁，然后尝试（非阻塞地）锁住后面的锁；如果发现某一把锁暂时拿不到，就把已经拿到手的锁全部释放掉，稍等一下，再重新按顺序尝试一遍，如此反复，直到有一次能够把所有需要的锁都成功地、连续地拿到手为止。因为这套流程里"要么全部拿到，要么一把都不留地放弃后重试"，就不会出现"线程1拿了一半、线程2也拿了一半，然后互相卡住"这种局面。

3.1 用流程图理解这套"要么全拿到，要么全放弃重试"的策略

从这张流程图能看出关键点：一旦发现无法凑齐所有需要的锁，宁可把已经到手的锁也主动放弃，回到起点重新来一遍，绝不会出现"抓着一把不放，同时死等另一把"的情况，这正是避免死锁的核心思路。

4. 完整代码示例：用 scoped_lock 修复之前的死锁问题

回顾前面死锁那一节的例子：两个资源，两把锁，两个线程以相反顺序申请锁，导致死锁。现在我们用 std::scoped_lock 把这个问题彻底解决掉：

#include <iostream>
#include <thread>
#include <mutex>
#include <chrono>
using namespace std::chrono_literals;
std::mutex resource1_mtx; // 保护资源1的锁
std::mutex resource2_mtx; // 保护资源2的锁
// 线程1：用scoped_lock同时申请资源1和资源2的锁，传入顺序是"资源1, 资源2"
void thread1_task() {
    std::cout << "线程1: 准备同时获取两把锁" << std::endl;
    // 一行代码安全地拿到两把锁，不管另一个线程传入的顺序是什么，都不会死锁
    std::scoped_lock lock(resource1_mtx, resource2_mtx);
    std::cout << "线程1: 已获得两把锁，开始处理资源" << std::endl;
    std::this_thread::sleep_for(50ms); // 模拟处理耗时
    // 离开这个函数时，lock对象被销毁，两把锁会被自动释放，不需要手动unlock
}
// 线程2：同样用scoped_lock，但传入顺序故意写成相反的"资源2, 资源1"
void thread2_task() {
    std::cout << "线程2: 准备同时获取两把锁" << std::endl;
    // 即使这里传入锁的顺序和线程1相反，scoped_lock内部的算法依然能保证不会死锁
    std::scoped_lock lock(resource2_mtx, resource1_mtx);
    std::cout << "线程2: 已获得两把锁，开始处理资源" << std::endl;
    std::this_thread::sleep_for(50ms);
}
int main() {
    std::thread t1(thread1_task);
    std::thread t2(thread2_task);
    t1.join();
    t2.join();
    // 这一次，程序一定能正常执行到这一行，不会像之前手动加锁那样卡死
    std::cout << "两个线程都执行完毕，没有发生死锁" << std::endl;
    return 0;
}

关键点解析：

• 注意这段代码里，thread1_task 传入锁的顺序是 (resource1_mtx, resource2_mtx)，而 thread2_task 传入的顺序是 (resource2_mtx, resource1_mtx)——顺序依然是相反的，如果换成之前手动 lock()/unlock() 的写法，这种相反的顺序正是导致死锁的根源。但因为这里用的是 std::scoped_lock，它内部的获取算法能够正确处理这种"顺序不一致"的情况，不会产生死锁。
• std::scoped_lock lock(resource1_mtx, resource2_mtx); 这一行，构造函数接受任意数量的互斥量作为参数（这里是两个），构造完成时，这些锁要么全部成功获取，要么这行代码本身会一直"忙"着重试，直到全部获取成功为止，不存在"只拿到一半"然后卡住等待的中间状态。
• 函数结束、lock 对象离开作用域时，它会自动把持有的所有锁依次释放，不需要写任何 unlock()，这一点和 lock_guard 的行为是一致的，只是 scoped_lock 能同时管理多把锁。

4.1 与手动加锁场景的对比时序图

5. std::lock_guard 与 std::scoped_lock 的对比

---

对比维度	std::lock_guard	std::scoped_lock
引入版本	C++11	C++17
能包装的锁数量	只能1把	0把或多把
多把锁时是否需要担心死锁	需要自己保证加锁顺序一致	内部算法自动避免死锁，不用担心传入顺序
典型用法	std::lock_guard<std::mutex> lock(mtx);	std::scoped_lock lock(mtx1, mtx2, ...);
支持adopt_lock	支持	支持

---

需要提一句，std::scoped_lock 和 std::unique_lock（下一节会详细介绍）虽然同属于 RAII 风格的锁包装器，但 std::unique_lock 和 lock_guard 一样，一次只能包装一把锁，只不过它比 lock_guard 提供了更灵活的操作（比如可以中途手动释放、延迟加锁等），这部分留到后面单独讲解。

小结

• std::scoped_lock 是 C++17 引入的 RAII 锁包装器，和 lock_guard 的最大区别是它可以同时包装0把或多把互斥锁。
• 它内部使用的加锁算法遵循"要么全部拿到、要么全部放弃重试"的策略，即便不同线程传入多把锁的顺序不一致，也能有效避免死锁，这正是它比手动 lock()/unlock() 更安全的核心原因。
• 一行 std::scoped_lock lock(mtx1, mtx2); 等价于手动写 std::lock(mtx1, mtx2); 加上两个使用 std::adopt_lock 的 lock_guard，scoped_lock 把这个组合彻底简化成了一行代码。
• 当一个任务需要同时持有多把锁时，应该优先考虑使用 std::scoped_lock，而不是手动管理多把 std::mutex 的加锁顺序，能大幅降低写出死锁 bug 的概率。

std::shared_lock：允许多个线程同时"只读"访问的锁

一、先想清楚一个问题：读操作真的需要互斥吗

前面讲的 std::mutex、std::lock_guard、std::unique_lock，保护的都是独占模式（exclusive mode）的访问：不管线程要对共享数据做什么操作，同一时刻永远只允许一个线程进去，其他线程一律排队等待。
但仔细想想，很多真实场景里，共享数据被访问的方式其实分成两种：读和写。如果多个线程都只是读取数据、不做任何修改，它们之间其实并不会互相干扰——十个人同时翻看同一本书的内容，并不会把书翻乱。真正需要"独占"的，只有写入这个动作：一旦有人要修改数据，就必须确保这一刻没有别的线程在读、也没有别的线程在写，不然读到的数据可能是修改到一半的、不完整的状态。
如果所有的读操作都用普通 std::mutex 去保护，效果就是：即使是十个只读的线程，也得排成一队一个一个来，完全没必要地牺牲了并发性能。std::shared_lock 搭配 std::shared_mutex（C++17 引入），就是专门用来解决这个问题的——它把访问模式区分成了共享模式和独占模式两种。
用一张图理解这两种模式的区别：

二、std::shared_lock 和 std::unique_lock 的核心区别

std::shared_lock 跟 std::unique_lock 一样，都是"通用的 mutex 所有权包装器"，也都支持延迟加锁（std::defer_lock）、转移锁的所有权这些特性，用法上非常相似。它们两者最关键的区别只有一点：
$$\begin{gathered} \text{std::unique\_lock：以「独占模式」获取和释放它包装的 mutex} \\ \text{std::shared\_lock：以「共享模式」获取和释放它包装的 mutex} \end{gathered}$$
也就是说，同一把支持共享模式的互斥量（std::shared_mutex），可以：

• 被一个 std::unique_lock 以独占模式持有（这时候任何其他线程都不能读也不能写，必须排队等待）；
• 或者被多个 std::shared_lock 同时以共享模式持有（这些线程都只能进行只读操作，互相之间不冲突，但只要有任何一个共享锁还在，独占模式的写操作就必须等它们都释放完）。
  用表格总结一下这两者的关系：

---

包装器	对应的锁定模式	同一时刻能有多少个线程持有	典型用途
std::unique_lock	独占模式（exclusive）	最多 1 个	需要修改共享数据的写操作
std::shared_lock	共享模式（shared）	可以有多个	只读取、不修改共享数据的读操作

---

需要注意的是，std::shared_lock 和 std::unique_lock 包装的必须是同一把、支持共享语义的 mutex，也就是 std::shared_mutex（或者 std::shared_timed_mutex），普通的 std::mutex 并不支持共享模式，没法配合 std::shared_lock 使用。

三、完整代码：多个读者 + 一个写者

下面用一个"共享数据由多个读线程和一个写线程共同访问"的例子，演示 std::shared_lock 和 std::unique_lock 是怎么配合同一把 std::shared_mutex 工作的。

#include <iostream>
#include <syncstream>  // 提供 std::osyncstream，避免多线程打印内容交叉错乱
#include <shared_mutex> // 提供 std::shared_mutex 和 std::shared_lock，C++17 起可用
#include <thread>
#include <vector>
#include <chrono>
#define sync_cout std::osyncstream(std::cout)
std::shared_mutex data_mutex; // 支持共享/独占两种模式的互斥量
int shared_data = 0;          // 被多个线程共同访问的数据
// 读线程：只读取数据，不修改，多个读线程可以同时进行
void reader(int id) {
    for (int i = 0; i < 3; ++i) {
        // 关键点：std::shared_lock 以"共享模式"获取 data_mutex，
        // 只要当前没有任何线程持有独占锁（也就是没有写操作正在进行），
        // 多个读线程调用这一行都能同时成功，互不阻塞。
        std::shared_lock<std::shared_mutex> lock(data_mutex);
        sync_cout << "读线程 " << id << " 读到的值: " << shared_data << "\n";
        std::this_thread::sleep_for(std::chrono::milliseconds(50)); // 模拟读取耗时
        // lock 离开作用域时自动释放这次共享锁
    }
}
// 写线程：需要独占访问，修改数据的时候不允许任何读线程或其他写线程同时介入
void writer() {
    for (int i = 0; i < 2; ++i) {
        std::this_thread::sleep_for(std::chrono::milliseconds(80)); // 模拟写之前的准备工作
        // 关键点：std::unique_lock 以"独占模式"获取 data_mutex，
        // 必须等所有正在进行的共享读锁都释放完毕，这次独占申请才会成功，
        // 一旦拿到，其他任何读锁或写锁都无法在此期间被获取。
        std::unique_lock<std::shared_mutex> lock(data_mutex);
        ++shared_data; // 安全地修改共享数据，此刻没有任何其他线程能同时访问它
        sync_cout << "写线程 将值修改为: " << shared_data << "\n";
        // lock 离开作用域时自动释放这次独占锁，其他线程才有机会继续访问
    }
}
int main() {
    std::vector<std::thread> readers;
    for (int i = 1; i <= 3; ++i) {
        readers.emplace_back(reader, i); // 创建 3 个读线程
    }
    std::thread writerThread(writer);    // 创建 1 个写线程
    for (auto& r : readers) {
        r.join();
    }
    writerThread.join();
    std::cout << "最终数据值: " << shared_data << std::endl;
    return 0;
}

代码逐段讲解：

• std::shared_mutex data_mutex;：这是本例中真正被保护的那把锁，它同时支持"共享模式加锁"（对应 lock_shared() / unlock_shared()）和"独占模式加锁"（对应普通的 lock() / unlock()），shared_lock 和 unique_lock 分别用的就是这两套接口。
• reader 函数里用 std::shared_lock<std::shared_mutex> lock(data_mutex); 获取共享锁：只要此刻没有写线程正在独占访问，多个读线程可以同时通过这一行，各自打印出当前看到的 shared_data 值，互相之间不会阻塞。
• writer 函数里用 std::unique_lock<std::shared_mutex> lock(data_mutex); 获取独占锁：这一步必须等所有当前持有共享锁的读线程都执行完各自的读取、释放锁之后，才能真正拿到锁，拿到之后，任何读线程或者其他写线程都无法在这段时间内访问 shared_data，从而保证 ++shared_data; 这个修改动作是完全独占、安全的。
• 因为读线程和写线程用的是不同的锁定模式（shared_lock vs unique_lock），底层这把 shared_mutex 会自动协调好"多个读者可以共存，但读者和写者、写者和写者之间必须互斥"这套规则，不需要开发者自己再额外去写判断逻辑。

3.1 用时序图看读写线程是怎么协调的

四、什么时候该考虑用 shared_lock + shared_mutex

std::shared_lock 和 std::shared_mutex 组合最适合的场景，是读操作远比写操作频繁的共享数据结构，比如一份很少变动、但会被大量线程反复查询的配置信息、缓存数据、或者只读的查找表。在这类场景下，把所有访问都用普通 std::mutex 保护会造成不必要的串行化，读线程之间明明互不冲突，却也要排队，浪费了并发的潜力；改用共享/独占两种模式区分开之后，大量的读操作可以真正并发执行，只有相对少见的写操作才需要独占等待，整体吞吐量通常会有明显提升。
反过来，如果读写操作的频率差不多，或者写操作占比也不低，shared_mutex 额外维护"共享/独占"两种状态的开销，可能会抵消掉并发读取带来的收益，这时候用普通 std::mutex 反而更简单、开销也更小，需要根据实际的读写比例来权衡。

五、小结

std::shared_lock 和 std::unique_lock 在用法上几乎是一对"孪生兄弟"，都支持延迟加锁、转移所有权这些通用能力，唯一的本质区别是它们对底层 mutex 采取的锁定模式不同：unique_lock 走的是独占模式，同一时刻只允许一个线程持有；shared_lock 走的是共享模式，允许多个只读线程同时持有。两者配合同一把 std::shared_mutex 使用，就能实现"多个读者可以并发访问，但只要有写者介入就必须让所有人都让路"这种经典的"读写锁"模式，特别适合读多写少的共享数据场景。理解了这一整套 mutex 包装器之后，下一步要介绍的是另一种同步机制——条件变量，它解决的是"线程之间互相等待、互相通知"这类需求。

条件变量（Condition Variables）详解

前面学的 std::mutex、std::lock_guard、std::scoped_lock 解决的都是"如何安全地访问共享数据"这个问题。但多线程编程里还有另外一类需求：一个线程需要等待某个条件成立，而这个条件是由另一个线程负责改变的——也就是线程之间需要"打招呼、发通知"。这就是**条件变量（Condition Variable）**要解决的问题。

1. 什么是条件变量

条件变量是 C++ 标准库提供的又一种同步原语，它让多个线程之间能够相互通信：一个或多个线程可以"等待"某个通知，而另一个线程可以在合适的时机"发出"这个通知，把等待中的线程唤醒。
条件变量永远要和一把互斥锁配合使用，不能单独存在。原因很直观：条件变量要检查的"条件"（比如"某个计数器是不是等于10了"）本身通常依赖一份共享数据，而共享数据的读写都需要锁来保护，条件变量只是在这份被锁保护的数据基础上，额外提供了"等待"和"通知"的能力。

2. 一个具体的例子：等待计数器变化

设想这样一个场景：有一个共享的计数器 counter，一个线程负责每隔一段时间就把它加一（一共加20次），另外还有两个线程都在"等待"这个计数器达到某种状态——一个等它"不再是0"，另一个等它"正好等于10"。
要实现"等待"这件事，其实有两种思路，下面这段完整代码里把这两种思路都写出来做对比：

#include <chrono>              // sleep_for 需要的时间单位
#include <condition_variable>  // std::condition_variable
#include <iostream>            // 标准输入输出
#include <mutex>               // std::mutex、std::lock_guard、std::unique_lock
#include <thread>              // std::thread
#include <vector>
int counter = 0; // 被多个线程共享的计数器
int main() {
    using namespace std::chrono_literals;
    std::mutex mtx;                 // 保护counter的锁
    std::mutex cout_mtx;             // 保护std::cout的锁，避免多线程打印内容交错
    std::condition_variable cv;      // 条件变量，用来通知/等待counter的变化
    // 任务1：负责每隔100毫秒把counter加一，一共加20次，每次加完就通知一下
    auto increment_counter = [&] {
        for (int i = 0; i < 20; ++i) {
            std::this_thread::sleep_for(100ms); // 模拟耗时的工作
            mtx.lock();
            ++counter;      // 修改共享数据前先加锁
            mtx.unlock();   // 修改完立刻解锁
            // 通知一个正在等待cv的线程："我刚才改了counter，你可以来检查一下了"
            cv.notify_one();
        }
    };
    // 任务2：用"手动轮询"的方式等待counter变得不是0（不使用条件变量）
    auto wait_for_counter_non_zero_mtx = [&] {
        mtx.lock();
        while (counter == 0) {
            // 每次发现counter还是0，就先把锁释放掉，睡一会儿，再重新加锁检查
            // 这种"反复加锁-检查-解锁-睡眠"的方式叫做"忙等"或"轮询"
            mtx.unlock();
            std::this_thread::sleep_for(10ms);
            mtx.lock();
        }
        mtx.unlock();
        std::lock_guard<std::mutex> cout_lck(cout_mtx);
        std::cout << "Counter is non-zero" << std::endl;
    };
    // 任务3：用条件变量的方式等待counter变得正好等于10
    auto wait_for_counter_10_cv = [&] {
        // unique_lock比lock_guard更灵活，因为cv.wait内部需要能够临时释放/重新获取这把锁
        // (lock_guard不支持中途解锁，所以这里必须用unique_lock)
        std::unique_lock<std::mutex> lck(mtx);
        // cv.wait的第二个参数是一个"判断条件是否成立"的函数(这里是lambda)
        // 只要这个lambda返回true，wait就会返回，继续往下执行
        // 如果返回false，wait内部会自动释放lck持有的锁，让线程进入休眠等待状态
        cv.wait(lck, [] { return counter == 10; });
        // 走到这里，说明条件已经成立(counter恰好等于10)，
        // 而且lck这把锁在wait返回时，已经被自动重新加锁了
        std::lock_guard<std::mutex> cout_lck(cout_mtx);
        std::cout << "Counter is: " << counter << std::endl;
    };
    // 创建三个线程，分别执行上面定义的三个任务
    std::thread t1(wait_for_counter_non_zero_mtx);
    std::thread t2(wait_for_counter_10_cv);
    std::thread t3(increment_counter);
    t1.join();
    t2.join();
    t3.join();
    return 0;
}

3. 第一种等待方式：手动轮询（wait_for_counter_non_zero_mtx）

这个函数的逻辑是：先加锁，检查 counter 是不是 0，如果是 0，就解锁、睡 10 毫秒，然后再重新加锁检查一次，如此循环，直到 counter 不再是 0 为止。
这种写法能工作，但有明显的缺点：

• 线程要不停地"醒来检查、发现不满足、又睡回去"，即使条件迟迟不满足，这个循环也会持续不断地被唤醒、加锁、检查、解锁，白白消耗 CPU 资源，这种模式通常被称为忙等（busy waiting）或者轮询（polling）。
• 响应不够及时：因为是每隔 10 毫秒才检查一次，条件实际满足的那一刻，和线程真正"发现"条件满足的那一刻之间，可能存在最多 10 毫秒的延迟。

4. 第二种等待方式：条件变量（wait_for_counter_10_cv）

条件变量的写法能大大简化这类"等待某个条件成立"的代码，而且效率更高——线程在条件不满足的时候，是真正进入"休眠"状态的，不会占用 CPU 去反复检查，只有等被明确"通知"到了，才会醒过来再检查一次。
cv.wait(lck, predicate) 这一行代码内部具体做的事情，可以拆解成下面几步：

1. 检查 predicate()（也就是传入的判断条件的lambda）是否为 true。如果已经是 true 了，wait 直接返回，不需要等待。
2. 如果 predicate() 是 false，wait 会自动释放 lck 持有的锁，然后让当前线程进入休眠等待状态（这一步很关键：如果不释放锁，其他线程根本没办法去修改 counter，那条件永远也不可能变成 true）。
3. 当有别的线程调用了 cv.notify_one() 或 cv.notify_all()，休眠中的线程会被唤醒，重新获取 lck 这把锁，再次检查 predicate() 是否为 true。
4. 如果这次检查依然是 false，重复步骤2和3继续等待；如果是 true，wait 函数返回，此时 lck 已经处于加锁状态，可以安全地继续访问被保护的共享数据（这里就是 counter）。
   这也解释了为什么这里必须用 std::unique_lock 而不是 std::lock_guard：lock_guard 一旦构造就一直持有锁，直到析构才释放，中途没办法主动解锁；而 cv.wait 内部需要能够"先释放锁进入休眠，被唤醒后再重新加锁"，这种"中途可以解锁、之后还能重新加锁"的灵活性，只有 unique_lock 才具备。

4.1 条件变量等待与通知的时序图

5. 一个必须注意的问题：虚假唤醒（Spurious Wakeup）

条件变量有一个特殊的现象需要特别注意：有时候，一个正在等待的线程可能会被莫名其妙地唤醒，即使根本没有任何线程调用 notify_one() 或 notify_all()，这种现象叫做虚假唤醒（spurious wakeup）。这是操作系统底层实现的一种特性，属于标准允许的行为，不是bug。
正因为存在虚假唤醒的可能，条件变量在被唤醒之后，绝对不能想当然地认为"条件一定已经满足了"，必须重新检查一遍条件是否真的成立。这也正是为什么 cv.wait(lck, predicate) 要求传入一个 predicate（判断条件的函数）——wait 内部会在每次被唤醒时都重新调用一次这个 predicate，只有它返回 true，wait 才会真正返回；如果因为虚假唤醒被叫醒了一次，但 predicate() 检查发现条件其实还不成立，wait 会自动让线程重新进入休眠状态，继续等待下一次真正的通知。
用一个简单的等价关系表示 cv.wait(lck, predicate) 这种带谓词版本的内部逻辑：
$$\text{cv.wait(lck, predicate)}\equiv \text{while (!predicate()) { cv.wait(lck); }}$$
也就是说，带谓词参数的 wait 版本，本质上就是自动帮你写好了一个 while 循环，每次被唤醒（不管是正常通知还是虚假唤醒）都重新检查一次条件，直到条件真正满足才跳出循环，这样就从根本上避免了虚假唤醒带来的错误。

6. 通知的两种方式：notify_one() 与 notify_all()

条件变量提供了两个函数，用来"叫醒"正在等待的线程：

函数	作用
cv.notify_one()	只唤醒众多等待线程中的其中一个(具体哪一个由实现决定，不保证顺序)
cv.notify_all()	唤醒所有正在等待这个条件变量的线程，被唤醒的线程会各自重新检查条件是否成立

---

在本节的例子里，increment_counter 每次修改完 counter 都调用了 cv.notify_one()，因为每次只有 counter 的值发生了变化，通知一个等待中的线程去检查一下就够了。如果有多个线程都在等待同一个条件变量，并且这次条件的变化可能同时满足好几个线程各自的等待需求，就应该用 notify_all()，确保所有相关的等待线程都有机会被唤醒重新检查。

7. 两种等待方式的对比

---

对比维度	手动轮询(mutex+循环+sleep)	条件变量(condition_variable)
CPU占用	较高，需要反复被唤醒检查	较低，条件不满足时线程真正休眠
响应及时性	受轮询间隔限制，可能有延迟	一旦被notify，几乎立刻重新检查
代码复杂度	需要自己写循环、控制加锁解锁节奏	逻辑封装在wait内部，代码更简洁
是否需要考虑虚假唤醒	不涉及此问题	需要，但带谓词的wait已经自动处理好了

---

小结

• 条件变量用于实现线程之间"等待通知"的协作机制，并且必须搭配一把互斥锁一起使用，因为条件变量所依赖的判断条件通常建立在需要被保护的共享数据之上。
• 相比手动用"加锁-检查-解锁-睡眠"的轮询方式等待条件成立，条件变量能让线程在条件不满足时真正进入休眠，减少不必要的CPU占用，并且能在条件变化时更及时地被唤醒。
• cv.wait(lck, predicate) 要求传入 std::unique_lock（而不是 lock_guard），因为等待过程中需要反复地"释放锁进入休眠、被唤醒后重新加锁"，这种灵活性只有 unique_lock 才提供。
• 虚假唤醒是条件变量固有的一种现象，线程被唤醒后不能假设条件一定成立，必须重新检查；带谓词参数的 wait 版本已经自动处理好了这个重新检查的循环逻辑。
• 通知有两种方式：notify_one() 只唤醒一个等待线程，notify_all() 唤醒所有等待线程，应该根据实际场景（条件变化是否可能同时满足多个等待者）来选择使用哪一个。
  下一节会在条件变量的基础上，用一把互斥锁加两个条件变量，实现一个真正实用的组件——线程安全的队列。

实现一个多线程安全队列：环形缓冲区 + 条件变量

一、为什么线程之间常常需要一个"队列"

多个线程之间要传递数据，一种非常经典的方式就是用队列（先进先出，FIFO）。举个具体场景：假设需要从网络连接里尽快接收数据包，一个线程可能忙不过来"既要收包、又要处理包"，这时候常见的做法是：一个线程专门负责接收数据包，另一个线程专门负责处理数据包，两者之间用一个队列传递数据——前一个线程把收到的包塞进队列（生产者），后一个线程从队列里取出包来处理（消费者）。
这种"一个生产者、一个消费者"的模式叫 SPSC（Single-Producer-Single-Consumer），是最简单的一种情况，只需要保证顺序，数据从队列里读出来的顺序，一定和放进去的顺序一致。当然，实际场景也可能需要多个生产者（多个数据来源）或者多个消费者（单个消费者处理能力不够），这一节要实现的队列会兼顾这几种情况都能正常工作，但先以最简单的"一生产者、一消费者"来理解核心逻辑。
用一张图理解队列在这里扮演的角色：

二、设计思路：用"环形缓冲区"实现一个固定容量的队列

2.1 为什么选环形缓冲区

要存放队列里的数据，可以选择链表（不限容量，但每次插入删除都要动态分配内存）、std::queue（同样不限容量）等方式，但这一节选择了一种更"轻量"的实现：用 std::vector<T> 作为底层存储，容量在构造时就固定下来，之后不再发生任何内存分配。这样做的好处是：内存只在构造的时候统一申请一次，后续频繁的 push/pop 操作都不涉及内存分配，性能更稳定、更可预测。
这个固定大小的 vector 会被当成环形缓冲区（ring buffer）来使用：数据写到数组末尾之后，下一个数据会"绕回"到数组开头继续写，读取也是同样的道理，整个数组的头尾在逻辑上被"接"成了一个环。
用 ASCII 图表示这种"环"的结构（假设容量是 6 个格子）：

                     索引 0
                   +-------+
           索引 5   |       |   索引 1
         +---------+       +---------+
         |                           |
         |         环形缓冲区          |
         |         （容量 6）          |
         |                           |
         +---------+       +---------+
           索引 4   |       |   索引 2
                   +-------+
                     索引 3

用两个下标 head_ 和 tail_ 分别记录"下一次读取应该从哪个位置开始"和"下一次写入应该写到哪个位置"，当某一个下标走到数组末尾时，再往前走一步就直接"绕回"到下标 0，这个"绕回"的动作，用取模运算就能优雅地实现：
$$\text{next\_index}=(\text{当前下标}+1)\mathrm{mod}\text{capacity}$$
取模运算的效果就是"除以容量取余数"，一旦下标加 1 之后达到了容量大小，取模就会自动把它折回 0，天然实现了环形绕回的效果，不需要写额外的 if 判断。

2.2 怎么判断队列是空的还是满的

head_ 指向下一次要读取的位置，tail_ 指向下一次要写入的位置。这里采用一种巧妙但需要注意的约定：
$$\begin{gathered} \text{队列为空}⟺\text{tail\_}=\text{head\_} \\ \text{队列已满}⟺(\text{tail\_}+1)\mathrm{mod}\text{capacity}=\text{head\_} \end{gathered}$$
也就是说：如果头尾指针重合，说明队列是空的（还没有任何等待读取的数据）；如果尾指针"再往前走一步"就会追上头指针，说明队列已经写满了，不能再继续写了。
这种判断方式有一个值得留意的代价：真正能用的容量，会比构造时传入的 capacity 少 1 个格子。原因是如果允许把所有格子都填满（也就是 tail_ 真的追上了 head_），那"队列已满"和"队列为空"这两种状态在头尾指针的表现上就会完全一样（都是 tail_ == head_），没法区分开来。为了避免这种二义性，这里选择"预留一个格子不用"，当只剩最后一个格子空着的时候，就已经把队列判定为"满"了，宁可少用一格，也要保证空和满这两种状态能被准确区分。
用表格总结一下这个设计上的取舍：

判断方式	需要额外维护的状态	真实可用容量
头尾指针重合判断空，差一格判断满（本节采用）	不需要额外的标志位或计数器	capacity - 1
额外维护一个"是否为满"标志位，或者维护一个计数器	需要一个额外的共享变量	capacity

---

之所以选择第一种、"损失一格容量"的方式，是因为这一节的设计目标是尽量减少多个线程之间共享和读写的数据。如果额外加一个"是否已满"的标志位或者计数器，这个变量就会被生产者线程和消费者线程同时读写，等于又多了一份需要小心保护的共享状态；宁可牺牲一格容量，换来"不需要再多维护一份共享标志"的简单和安全，这在后面的实现里会看到，也是为将来更进一步优化（用不需要锁的方式重新实现这个队列）打基础的思路。

三、队列类的基础框架

先把类的骨架和几个基础的辅助函数搭出来：

#include <vector>
#include <cstddef> // 提供 std::size_t
template <typename T>
class synchronized_queue {
public:
    // 构造函数：capacity_ 记录队列的容量，buffer_ 在构造时一次性分配好所有空间
    explicit synchronized_queue(std::size_t size)
        : capacity_{size}, buffer_(capacity_) {
    }
private:
    // 关键点：计算某个下标"往前走一步"之后的位置，用取模运算实现环形绕回，
    // 如果 index 已经是最后一个下标（capacity_ - 1），加 1 取模后会变回 0。
    std::size_t next(std::size_t index) const {
        return (index + 1) % capacity_;
    }
    // 关键点：如果 tail_ 再往前走一步就会等于 head_，说明只剩最后一个空格了，
    // 按照本节的约定，这种情况就判定为"队列已满"。
    bool is_full() const {
        return next(tail_) == head_;
    }
    // 关键点：头尾指针相等，说明没有任何待读取的数据，队列为空
    bool is_empty() const {
        return tail_ == head_;
    }
    std::size_t head_{0};   // 下一次读取（pop）操作应该读取的下标
    std::size_t tail_{0};   // 下一次写入（push）操作应该写入的下标
    std::size_t capacity_;  // 队列的容量（构造时固定，之后不变）
    std::vector<T> buffer_; // 真正存放数据的底层数组，充当环形缓冲区
};

这几个辅助函数虽然简单，但是后面 push、pop 的所有逻辑都建立在它们之上：next() 负责"往前挪一格并处理绕回"，is_full() 和 is_empty() 负责判断当前能不能继续写入或者读取。

四、push 和 pop：用条件变量协调"队列满了"和"队列空了"

4.1 为什么需要条件变量

push（生产者插入数据）和 pop（消费者取出数据）都有一个共同的诉求：如果条件不满足，就要等待，而不是直接出错或者空转检查。

• 队列满了的时候，生产者线程应该停下来等，一直等到消费者取走了至少一个元素、腾出空位为止，才能继续插入。
• 队列空了的时候，消费者线程应该停下来等，一直等到生产者放入了至少一个元素，才能继续取出。
  这正是条件变量（std::condition_variable）最擅长解决的问题：让一个线程在"条件不满足"时安静地睡眠等待，不占用 CPU，一旦另一个线程改变了条件、并且发出通知，等待的线程才会被唤醒继续检查。
  给队列类加上这几个成员：

#include <mutex>
#include <condition_variable>
// ……
std::mutex mtx_;                       // 保护 head_、tail_、buffer_ 这些共享状态的互斥量
std::condition_variable not_full_;     // 生产者在队列满时，在这个条件变量上等待
std::condition_variable not_empty_;    // 消费者在队列空时，在这个条件变量上等待

之所以要用两个条件变量而不是一个，是因为生产者和消费者等待的"条件"根本不是同一件事：生产者等的是"队列不满了"，消费者等的是"队列不空了"，把它们分开，可以让"通知"更精准——生产者放入一个元素之后，只需要唤醒可能在等"队列不空"的消费者，不需要去打扰同样在等待、但等的是完全不同条件的其他生产者。

4.2 push 函数的实现

void push(const T& item) {
    // 关键点：条件变量的 wait() 要求配合 std::unique_lock 使用，
    // 而不能用 std::lock_guard，因为 wait() 内部需要"临时释放锁、
    // 睡眠等待、被唤醒后重新加锁"这一整套流程，lock_guard 不支持临时解锁。
    std::unique_lock<std::mutex> lock(mtx_);
    // 关键点：wait 的第二个参数是一个"谓词"（返回 bool 的可调用对象），
    // 只有当 !is_full() 为 true（也就是队列不满）时，wait 才会真正往下走，
    // 否则会一直阻塞在这里，并且在阻塞期间自动释放 mtx_，
    // 让消费者线程有机会拿到锁去执行 pop、腾出空位。
    //
    // 这里必须用带谓词的 wait，而不是"无条件等一次通知就往下走"，
    // 是因为条件变量存在"虚假唤醒"（spurious wakeup）的可能——
    // 也就是说，即使没有任何线程真的调用了 notify，
    // 操作系统偶尔也可能无缘无故把等待中的线程唤醒。
    // 加了谓词之后，即使被虚假唤醒，wait 也会重新检查 is_full()，
    // 如果条件仍然不满足，会继续睡回去，不会被误判为"可以往下走了"。
    not_full_.wait(lock, [this] { return !is_full(); });
    // 走到这里，说明队列确实不满了，可以安全地写入新元素
    buffer_[tail_] = item;   // 把新元素写到当前 tail_ 指向的位置
    tail_ = next(tail_);     // tail_ 往前挪一格（环形绕回由 next() 负责）
    // 关键点：这里选择显式调用 unlock()，而不是让 unique_lock 在函数结束、
    // 自然析构时才释放锁，原因是紧接着要调用 notify_one()，
    // 如果这时候还占着锁，被唤醒的消费者线程醒来后第一件事就是尝试重新加锁，
    // 但锁还在生产者手里没释放，消费者只能白白多等一轮，效率上不划算，
    // 所以先手动解锁，再发通知，让被唤醒的线程能立刻拿到锁继续往下走。
    lock.unlock();
    // 关键点：告诉可能正在等待"队列不空"的消费者线程——
    // 现在队列里确实有新数据了，可以起来检查一下了。
    not_empty_.notify_one();
}

4.3 pop 函数的实现

pop 的逻辑跟 push 是对称的，只是角色和条件反过来了：

void pop(T& item) {
    std::unique_lock<std::mutex> lock(mtx_);
    // 关键点：只有队列不空的时候才能继续读取，
    // 否则在这里阻塞等待，并释放 mtx_，让生产者有机会写入新数据。
    not_empty_.wait(lock, [this] { return !is_empty(); });
    item = buffer_[head_]; // 把 head_ 指向的元素取出来，赋值给调用者传入的 item
    head_ = next(head_);   // head_ 往前挪一格
    lock.unlock(); // 同样是先解锁，再通知，避免被唤醒的线程白等一轮
    // 消费完一个元素之后，队列肯定不再是"满"的状态了，
    // 通知可能正在等待"队列不满"的生产者线程可以继续写入了。
    not_full_.notify_one();
}

4.4 用时序图梳理生产者和消费者的完整互动过程

假设消费者先启动，恰好赶上队列还是空的：

五、非阻塞版本：try_push 和 try_pop

push 和 pop 都是阻塞式的：条件不满足就一直等，不会返回。但有些场景下，线程可能不想傻等——比如希望在队列暂时满/空的时候，先去做点别的独立工作，过一会儿再回来重试。这就需要非阻塞版本的 try_push 和 try_pop。

bool try_push(const T& item) {
    // 关键点：std::try_to_lock 让这次加锁尝试变成非阻塞的，
    // 如果这把锁正被其他线程占着，这一行不会等待，会立刻返回，
    // lock 对象内部会记录"这次到底有没有真的拿到锁"。
    std::unique_lock<std::mutex> lock(mtx_, std::try_to_lock);
    // 关键点：这里要检查两件事——
    // 1. !lock：说明刚才没有真的抢到锁（转换成 bool 表示是否持有锁）；
    // 2. is_full()：即使抢到了锁，但队列已经满了，也不能继续写入。
    // 任意一个条件成立，都直接返回 false，把"要不要重试"的决定权交给调用者，
    // 而不是像 push 那样在这里死等。
    if (!lock || is_full()) {
        return false;
    }
    buffer_[tail_] = item;
    tail_ = next(tail_);
    lock.unlock();
    // 关键点：即使是非阻塞版本，插入成功之后依然要通知消费者，
    // 因为消费者完全可能调用的是阻塞版的 pop，仍然需要被正常唤醒。
    not_empty_.notify_one();
    return true; // 明确告诉调用者：这次插入成功了
}
bool try_pop(T& item) {
    std::unique_lock<std::mutex> lock(mtx_, std::try_to_lock);
    if (!lock || is_empty()) {
        return false;
    }
    item = buffer_[head_];
    head_ = next(head_);
    lock.unlock();
    not_full_.notify_one();
    return true;
}

try_push/try_pop 和 push/pop 最大的区别，就是把"该不该等待"这个决定，从队列内部移交给了调用者：普通版本自己在内部用条件变量死等，非阻塞版本只负责"能做就做，做不了立刻如实告知"，至于要不要重试、重试前要不要先做点别的事，完全由外部代码自己决定。
用表格对比一下这两组接口：

接口	条件不满足时的行为	是否需要条件变量参与等待
push / pop	阻塞，在条件变量上睡眠等待，直到条件满足	是
try_push / try_pop	立刻返回 false，不等待，交给调用者自行决定	否（只在成功后负责 notify）

---

六、完整可运行代码

把前面讲的所有部分拼在一起，补上一个 main 函数，用一个生产者线程和一个消费者线程演示整个队列的运作：

#include <iostream>
#include <mutex>
#include <condition_variable>
#include <vector>
#include <thread>
#include <chrono>
#include <syncstream>
#define sync_cout std::osyncstream(std::cout)
namespace async_prog {
template <typename T>
class queue {
public:
    explicit queue(std::size_t capacity)
        : capacity_{capacity}, buffer_(capacity) {
    }
    // 阻塞式插入：队列满时会一直等待，直到有空位
    void push(const T& item) {
        std::unique_lock<std::mutex> lock(mtx_);
        not_full_.wait(lock, [this] { return !is_full(); });
        buffer_[tail_] = item;
        tail_ = next(tail_);
        lock.unlock();
        not_empty_.notify_one();
    }
    // 非阻塞式插入：拿不到锁或队列已满，立刻返回 false
    bool try_push(const T& item) {
        std::unique_lock<std::mutex> lock(mtx_, std::try_to_lock);
        if (!lock || is_full()) {
            return false;
        }
        buffer_[tail_] = item;
        tail_ = next(tail_);
        lock.unlock();
        not_empty_.notify_one();
        return true;
    }
    // 阻塞式取出：队列空时会一直等待，直到有新数据
    void pop(T& item) {
        std::unique_lock<std::mutex> lock(mtx_);
        not_empty_.wait(lock, [this] { return !is_empty(); });
        item = buffer_[head_];
        head_ = next(head_);
        lock.unlock();
        not_full_.notify_one();
    }
    // 非阻塞式取出：拿不到锁或队列已空，立刻返回 false
    bool try_pop(T& item) {
        std::unique_lock<std::mutex> lock(mtx_, std::try_to_lock);
        if (!lock || is_empty()) {
            return false;
        }
        item = buffer_[head_];
        head_ = next(head_);
        lock.unlock();
        not_empty_.notify_one();
        return true;
    }
private:
    [[nodiscard]] std::size_t next(std::size_t idx) const noexcept {
        return (idx + 1) % capacity_;
    }
    [[nodiscard]] bool is_empty() const noexcept { return head_ == tail_; }
    [[nodiscard]] bool is_full() const noexcept { return next(tail_) == head_; }
    std::mutex mtx_;
    std::condition_variable not_empty_;
    std::condition_variable not_full_;
    std::size_t head_{0};
    std::size_t tail_{0};
    std::size_t capacity_;
    std::vector<T> buffer_;
};
} // namespace async_prog
int main() {
    // 容量设为 4，真正能同时存放的元素是 3 个（前面讲过会损失一格）
    async_prog::queue<int> q(4);
    // 生产者线程：依次把 0 到 9 这 10 个整数放进队列
    std::thread producer([&q] {
        for (int i = 0; i < 10; ++i) {
            q.push(i); // 如果队列满了，这里会自动阻塞等待，直到消费者腾出空位
            sync_cout << "生产者放入: " << i << "\n";
            std::this_thread::sleep_for(std::chrono::milliseconds(20));
        }
    });
    // 消费者线程：取出 10 个整数并打印
    std::thread consumer([&q] {
        for (int i = 0; i < 10; ++i) {
            int value;
            q.pop(value); // 如果队列空了，这里会自动阻塞等待，直到生产者放入新数据
            sync_cout << "消费者取出: " << value << "\n";
            std::this_thread::sleep_for(std::chrono::milliseconds(50)); // 消费者稍慢一些，更容易观察到队列变满
        }
    });
    producer.join();
    consumer.join();
    std::cout << "生产者和消费者都已完成工作" << std::endl;
    return 0;
}

关于这个演示的说明：代码里故意让消费者的处理速度比生产者慢（消费者每次多睡 30 毫秒），这样运行起来更容易观察到"生产者把队列写满、被迫等待消费者"这种阻塞效果——控制台上会看到"生产者放入"的输出间隔逐渐被"消费者取出"的节奏拖慢，正是 not_full_.wait(...) 在背后发挥作用的直观体现。

七、小结

这个队列的设计核心，是用环形缓冲区（固定大小的 vector，通过取模运算实现下标绕回）来存储数据，用头尾指针相对位置（而不是额外的标志位或计数器）判断队列的空和满，用两个独立的条件变量分别协调"生产者等待队列不满"和"消费者等待队列不空"这两种不同的等待需求。push/pop 是阻塞式接口，条件不满足就安静等待；try_push/try_pop 是非阻塞式接口，把"是否重试"的决定权交还给调用者。这一整套用互斥量加条件变量搭建起来的同步机制，正是构建更复杂并发工具（比如前面提到过的线程池）的基础构件。

C++20 信号量（Semaphore）从零详解

1. 什么是信号量

想象一个停车场只有 3 个车位。停车场入口有一块电子牌子，显示"剩余车位数"。每进去一辆车，数字就减一；每开走一辆车，数字就加一。当数字变成 0 时，后面来的车必须在门口排队等待，直到有车开走腾出车位。
信号量（Semaphore）就是计算机科学里的这块"电子牌子"。它本质上是一个计数器，配合两个原子操作：

• P 操作（wait / acquire，也叫申请资源）：把计数器减一。如果减完之后计数器小于 0（或者说计数器已经是 0，说明没有资源了），当前线程就会被阻塞，直到有资源被释放。
• V 操作（signal / release，也叫释放资源）：把计数器加一，并且如果有线程正在等待，就唤醒其中一个。
  用数学语言描述，设信号量的计数值为 $S$：
  $$P(S):S=S-1,\text{若}S<0\text{则阻塞当前线程}$$
  $$V(S):S=S+1,\text{若有等待线程则唤醒其中一个}$$
  信号量最初由荷兰计算机科学家 Dijkstra 提出，“P” 和 “V” 分别来自荷兰语 Proberen（尝试）和 Verhogen（增加）。

---

2. 二元信号量与计数信号量

信号量按照计数器能取的最大值，分为两种：

• 二元信号量（Binary Semaphore）：计数器只能是 0 或 1，概念上很像互斥锁（mutex）——要么"被占用"（0），要么"空闲"（1）。
• 计数信号量（Counting Semaphore）：计数器可以是任意非负整数（比如上面例子里的 3），用来控制"同时最多允许多少个线程访问某种有限数量的资源"。
  下面这个表格总结两者的区别：

---

类型	计数器范围	典型用途
二元信号量	0 或 1	保护单一临界区，类似互斥锁
计数信号量	0 到 N（N 为初始值）	限制同时访问某资源的线程数量，例如连接池、线程池

---

3. 信号量和互斥锁（mutex）的区别

虽然二元信号量看起来跟 mutex 很像，但它们有本质的不同：

对比项	互斥锁 mutex	二元信号量
所有权概念	有"所有权"，谁加锁谁必须解锁	没有所有权，任何线程都可以释放
加锁/解锁线程	必须是同一个线程	可以是不同线程（比如线程 A 等待，线程 B 释放）
典型场景	保护临界区（互斥访问）	线程间的信号通知（同步），比如生产者通知消费者
递归调用	部分实现支持递归锁	一般不支持递归获取

---

也就是说：mutex 主要解决"互斥"问题（同一时刻只有一个线程能进入临界区），而信号量除了能做互斥，还天然支持"通知/同步"——一个线程释放信号量、另一个线程去获取，这在生产者-消费者模型里非常常用。

4. C++20 中的信号量类型

C++20 在头文件 <semaphore> 中提供了两个类模板：

• std::counting_semaphore<LeastMaxValue>：计数信号量，模板参数 LeastMaxValue 表示计数器允许达到的最小上限值（编译期指定，默认是 INT_MAX 左右的一个很大的值）。
• std::binary_semaphore：其实就是 std::counting_semaphore<1> 的别名，计数器只能是 0 或 1。
  常用成员函数：
• acquire()：对应 P 操作，计数器减一；如果计数器已经是 0，则阻塞直到有资源可用。
• release(n = 1)：对应 V 操作，把计数器加 n，并唤醒相应数量的等待线程。
• try_acquire()：非阻塞尝试获取，成功返回 true，失败立刻返回 false。
• try_acquire_for(duration) / try_acquire_until(time_point)：带超时的尝试获取。

---

5. 代码示例一：用二元信号量模拟互斥锁

下面这个例子用 std::binary_semaphore 保护一个共享变量的自增操作，效果等价于用 mutex 加锁，但请注意——这里故意演示了"跨线程释放"这一 mutex 做不到的特性。

#include <iostream>   // 用于标准输入输出 std::cout
#include <semaphore>  // C++20 信号量头文件，提供 std::binary_semaphore
#include <thread>     // 用于创建线程 std::thread
#include <vector>     // 用于存放多个线程对象
// 全局二元信号量，初始值为 1，表示"资源空闲，可以进入临界区"
// 这里用 std::binary_semaphore 而不是 std::counting_semaphore<1>，两者等价，binary_semaphore 更直观
std::binary_semaphore g_binary_sem{1};
// 共享资源：一个普通的整数，多个线程会对它进行自增操作
long g_shared_counter = 0;
// 每个线程要执行的任务：循环自增共享计数器
void worker_increment(int loop_times)
{
    for (int i = 0; i < loop_times; ++i)
    {
        g_binary_sem.acquire();      // P 操作：申请进入临界区，如果计数器是 0 就阻塞在这里
        ++g_shared_counter;          // 临界区：安全地修改共享变量
        g_binary_sem.release();      // V 操作：释放信号量，计数器加一，唤醒等待的其他线程
    }
}
int main()
{
    const int thread_count = 4;      // 开 4 个线程
    const int loop_times = 100000;   // 每个线程自增 100000 次
    std::vector<std::thread> threads; // 用于存放所有线程对象，方便后面统一 join
    threads.reserve(thread_count);
    // 创建并启动 4 个线程，每个线程都跑 worker_increment 函数
    for (int i = 0; i < thread_count; ++i)
    {
        threads.emplace_back(worker_increment, loop_times);
    }
    // 等待所有线程执行完毕
    for (auto& t : threads)
    {
        t.join();
    }
    // 理论上最终结果应该是 4 * 100000 = 400000，如果结果不对说明同步出了问题
    std::cout << "最终计数结果: " << g_shared_counter << std::endl;
    std::cout << "期望结果: " << (thread_count * loop_times) << std::endl;
    return 0;
}

代码讲解：

1. std::binary_semaphore g_binary_sem{1}; 创建了一个初始计数值为 1 的二元信号量，“1” 表示一开始资源是可用的（类比 mutex 刚创建时是"未加锁"状态）。
2. g_binary_sem.acquire(); 是 P 操作。第一个到达的线程会把计数器从 1 减到 0 然后顺利通过；后面的线程再调用 acquire() 时，因为计数器已经是 0，就会被阻塞，进入等待队列。
3. 临界区里只做一件事：++g_shared_counter;，这一步因为有信号量保护，同一时刻只有一个线程能执行，所以不会发生数据竞争。
4. g_binary_sem.release(); 是 V 操作，把计数器加回 1，并唤醒一个正在等待的线程，让它有机会进入临界区。
5. 整个程序创建 4 个线程，各自循环 10 万次对同一个变量做自增，最后验证结果是否等于 $4\times 100000=400000$。如果没有信号量保护，多线程同时读写 g_shared_counter 会发生数据竞争，最终结果会小于期望值。
   编译运行方式（需要支持 C++20 的编译器，比如 GCC 11 及以上）：

g++ -std=c++20 -pthread semaphore_binary_demo.cpp -o semaphore_binary_demo
./semaphore_binary_demo

---

6. 代码示例二：用计数信号量限制并发数量（连接池场景）

计数信号量最典型的用法是限制"同时能有多少个线程访问某种有限资源"。下面模拟一个只有 3 个连接名额的"数据库连接池"，即使有 6 个线程同时想访问，也只能有 3 个同时在使用连接，其余的必须排队等待。

#include <iostream>   // 标准输入输出
#include <semaphore>  // C++20 信号量，提供 std::counting_semaphore
#include <thread>     // 多线程支持
#include <vector>     // 存放线程对象
#include <chrono>     // 用于模拟耗时操作 std::this_thread::sleep_for
#include <mutex>      // 仅用于保护 std::cout 的输出，防止打印内容交错
// 计数信号量，模板参数 3 表示"计数器至少能保证达到 3"（这是 LeastMaxValue，编译期常量）
// 初始化值 3 表示初始有 3 个连接名额可用
std::counting_semaphore<3> g_connection_pool_sem{3};
// 仅用来保护 std::cout，防止多个线程同时打印导致输出乱序交错（这是普通 mutex，不是本文重点）
std::mutex g_cout_mutex;
// 模拟每个线程去"借用连接、使用连接、归还连接"的过程
void access_database(int worker_id)
{
    {
        std::lock_guard<std::mutex> lock(g_cout_mutex);
        std::cout << "线程 " << worker_id << " 正在排队等待连接..." << std::endl;
    }
    g_connection_pool_sem.acquire();   // P 操作：申请一个连接名额，如果 3 个名额都被占用则阻塞
    {
        std::lock_guard<std::mutex> lock(g_cout_mutex);
        std::cout << "线程 " << worker_id << " 获得连接，开始使用数据库" << std::endl;
    }
    // 模拟使用数据库连接需要耗费一些时间
    std::this_thread::sleep_for(std::chrono::milliseconds(500));
    {
        std::lock_guard<std::mutex> lock(g_cout_mutex);
        std::cout << "线程 " << worker_id << " 使用完毕，归还连接" << std::endl;
    }
    g_connection_pool_sem.release();   // V 操作：归还连接名额，计数器加一，唤醒排队的线程
}
int main()
{
    const int total_workers = 6;       // 一共 6 个线程想要访问数据库
    std::vector<std::thread> threads;
    threads.reserve(total_workers);
    // 启动 6 个线程，但连接池只有 3 个名额，所以同一时刻最多只有 3 个线程能真正"使用数据库"
    for (int i = 1; i <= total_workers; ++i)
    {
        threads.emplace_back(access_database, i);
    }
    for (auto& t : threads)
    {
        t.join();
    }
    std::cout << "所有线程都已完成数据库访问" << std::endl;
    return 0;
}

代码讲解：

1. std::counting_semaphore<3> g_connection_pool_sem{3}; 中，尖括号里的 3 是模板参数 LeastMaxValue（表示计数器保证能达到的最小上限），花括号里的 3 是构造函数参数，表示信号量的初始计数值，即"一开始有 3 个连接名额"。
2. 每个线程先打印"排队中"，然后调用 acquire() 申请名额。前 3 个线程能立刻成功（计数器从 3 依次减到 0），后 3 个线程会被阻塞，直到前面有线程调用 release() 释放名额。
3. std::this_thread::sleep_for(...) 用来模拟"正在使用数据库"这个耗时操作，让效果更直观：运行程序会看到最多同时有 3 条"获得连接"的打印，其余线程要等前面线程 release() 之后才能继续。
4. g_connection_pool_sem.release(); 执行后，计数器加一，如果有线程在排队，会唤醒其中一个，让它去 acquire() 成功。
5. 这里额外用了一个普通的 std::mutex g_cout_mutex 仅仅是为了让多个线程打印时不会互相交错，这和信号量本身的知识点无关，只是为了让终端输出更清晰。
   编译运行方式：

g++ -std=c++20 -pthread semaphore_counting_demo.cpp -o semaphore_counting_demo
./semaphore_counting_demo

---

7. 时序图：信号量如何协调多个线程

下面用时序图展示"连接池"例子中，6 个线程与信号量之间的交互关系（为了简化，只画出 4 个线程）：

从图中可以看到：当计数器为 0 时新的申请者会被挂起，一旦有线程释放（release），信号量会立刻从等待队列里唤醒一个线程，让它继续获得资源，这就是信号量实现"限流"和"同步通知"的核心机制。

8. 信号量状态变化的直观示意（ASCII）

用 ASCII 图直观展示计数信号量（初始值为 3）在 6 个线程竞争下的计数变化过程：

初始状态：  [ 计数器 = 3 ]  可用名额：●●●
T1 acquire  ->  [ 计数器 = 2 ]  可用名额：●● 　　（T1 使用中）
T2 acquire  ->  [ 计数器 = 1 ]  可用名额：●  　　（T1 T2 使用中）
T3 acquire  ->  [ 计数器 = 0 ]  可用名额：　  　　（T1 T2 T3 使用中）
T4 acquire  ->  [ 计数器 = 0 ]  T4 阻塞等待...
T5 acquire  ->  [ 计数器 = 0 ]  T5 阻塞等待...
T6 acquire  ->  [ 计数器 = 0 ]  T6 阻塞等待...
T1 release  ->  [ 计数器 = 0 ]  唤醒 T4，T4 立刻获得名额
T2 release  ->  [ 计数器 = 0 ]  唤醒 T5，T5 立刻获得名额
T3 release  ->  [ 计数器 = 0 ]  唤醒 T6，T6 立刻获得名额
...

---

9. 总结

• 信号量本质是一个带阻塞能力的计数器，配合 P（申请，acquire）和 V（释放，release）两个操作使用。
• 二元信号量（std::binary_semaphore，即 std::counting_semaphore<1>）计数只能在 0 和 1 之间，概念上像 mutex，但没有"所有权"限制，可以跨线程释放，天然适合"通知/同步"场景。
• 计数信号量（std::counting_semaphore<N>）计数可以在 0 到 N 之间，适合限制"同时能访问某资源的线程数量"，例如连接池、限流器、生产者消费者模型中的缓冲区容量控制。
• C++20 之前，标准库没有内置信号量，开发者往往需要用 std::mutex + std::condition_variable 手动拼出等价功能；C++20 的 <semaphore> 头文件让这一切变得简单、标准、可移植。

计数信号量（Counting Semaphore）从零理解

1. 信号量到底是什么

信号量本质上就是一个带计数器的令牌桶。这个计数器可以在创建时被初始化成任意一个非负整数，表示"桶里有多少个令牌"。线程要使用共享资源之前，必须先从桶里拿一个令牌（这个动作叫 acquire）；用完之后，把令牌放回去（这个动作叫 release）。
用数学语言描述会更清楚。设信号量内部计数器为 $S$，初始值为 $S_0$：
$$\text{acquire()：}S\leftarrow S-1,\text{若此时 }S<0\text{ 则线程阻塞，直到有其他线程调用 release()}$$
$$\text{release()：}S\leftarrow S+1,\text{并唤醒一个正在等待的线程（如果有的话）}$$
也就是说，只要 $S>0$，acquire() 就可以立刻成功并把 $S$ 减一；一旦 $S$ 减到 $0$，下一个想 acquire() 的线程就必须等待，直到别的线程 release() 把 $S$ 加回正数。
std::counting_semaphore<> 是 C++20 才引入的标准库同步原语，和互斥锁（std::mutex）最大的区别是：

• 互斥锁：只有"锁住/没锁住"两种状态，本质上是计数器只能是 $0$ 或 $1$ 的特殊信号量（也叫二值信号量），而且谁加锁必须谁解锁（有"所有权"概念）。
• 计数信号量：计数器可以是任意非负整数，允许多个线程同时进入临界区；而且加锁和解锁可以是不同的线程（比如一个线程生产数据 release，另一个线程消费数据 acquire），这在协调"生产者-消费者"关系时非常自然。

2. 用信号量实现一个线程安全的环形队列

这一节的目标是：不用 std::condition_variable，改用两个计数信号量来实现一个多线程安全的环形缓冲队列（circular buffer queue）。

2.1 为什么需要两个信号量

对于一个容量固定的环形队列，任何时刻都存在两种"资源"：

• 空槽位（empty slot）：还没有被写入数据的位置，生产者需要空槽位才能写入。
• 满槽位（full slot）：已经写入了数据、还没被读走的位置，消费者需要满槽位才能读取。
  所以很自然地，用两个信号量分别表示这两种资源的数量：
• sem_empty_：表示"当前有多少个空槽位可用"，初始值等于队列容量 capacity_（因为一开始队列是空的，所有槽位都是空槽位）。
• sem_full_：表示"当前有多少个已经写好的槽位等待被读"，初始值为 $0$（因为一开始还没有任何数据）。
  用 ASCII 图直观表示一个容量为 6 的环形缓冲区，此刻已经写入了 3 个元素（head_ 指向下一个要读的位置，tail_ 指向下一个要写的位置）：

                tail_ (下一个写入位置)
                  |
                  v
    +---+---+---+---+---+---+
    | A | B | C |   |   |   |
    +---+---+---+---+---+---+
      ^
      |
    head_ (下一个读取位置)
  已写入(full) = 3   =>  sem_full_  当前计数 = 3
  空槽位(empty) = 3  =>  sem_empty_ 当前计数 = 3

head_ 和 tail_ 都是普通的下标，通过取模运算实现"环绕"：
$$\text{next(idx)}=(\text{idx}+1)\mathrm{mod}\text{capacity\_}$$

2.2 成员变量

template <typename T>
class queue {
    // 公开的方法（push、pop 等）和私有的辅助方法（next 等）省略
private:
    std::counting_semaphore<> sem_empty_;  // 空槽位计数：初始 = capacity_
    std::counting_semaphore<> sem_full_;   // 满槽位计数：初始 = 0
    std::size_t head_{ 0 };                // 下一个读取位置的下标
    std::size_t tail_{ 0 };                // 下一个写入位置的下标
    std::size_t capacity_;                 // 队列容量，用于取模实现环绕
    std::vector<T> buffer_;                // 真正存放数据的连续内存
};

要点：

• head_、tail_、capacity_ 都很直观，纯粹是为了在 std::vector 上模拟一个环。
• 这里没有 std::mutex，因为最初的设想是"只靠信号量就能同步"——但下面会看到这个想法有漏洞，最终还是要引入一把锁。

3. 第一版 push / pop（有漏洞的版本）

先看最朴素的写法，理解信号量的基本用法：

void push(const T& item) {
    sem_empty_.acquire();        // [1] 先占用一个"空槽位"名额；如果没有空槽位（计数为0），线程在此阻塞
    buffer_[tail_] = item;       // [2] 把元素写入 tail_ 指向的位置
    tail_ = next(tail_);         // [3] tail_ 前移一格（环绕）
    sem_full_.release();         // [4] 通知："又多了一个写好的槽位可供消费者读取"
}

逐行解释：

• [1] sem_empty_.acquire()：相当于数学式子里的 $S_{\text{empty}}\leftarrow S_{\text{empty}}-1$。如果队列已经满了（$S_{\text{empty}}=0$），这一步会让当前线程阻塞，直到消费者读走一个元素、sem_empty_.release() 被调用为止。
• [2][3]：真正把数据写进环形缓冲区，并把写指针 tail_ 移动到下一个位置。
• [4] sem_full_.release()：相当于 $S_{\text{full}}\leftarrow S_{\text{full}}+1$，告诉可能正在等待的消费者"现在多了一条数据可以读"。

void pop(T& item) {
    sem_full_.acquire();         // [1] 先占用一个"满槽位"名额；如果队列是空的（计数为0），线程在此阻塞
    item = buffer_[head_];       // [2] 读取 head_ 指向位置的数据
    head_ = next(head_);         // [3] head_ 前移一格（环绕）
    sem_empty_.release();        // [4] 通知："又多了一个空槽位可供生产者写入"
}

逻辑和 push 完全对称：先拿到"资格"（满槽位），再操作数据，最后归还另一种"资格"（空槽位）。

3.1 这版代码的问题

问题：sem_empty_ 只能保证"槽位数量够不够"，不能保证"同一时刻只有一个线程在改 tail_/buffer_"。
举个例子：如果队列容量是 10，当前有 2 个生产者线程同时调用 push。只要 sem_empty_ 的计数大于等于 2，两个线程都能顺利通过 [1] 这一步的 acquire()——因为信号量的计数只限制"总共能有多少个线程同时拿到令牌"，并不限制"这些线程会不会同时读写同一块内存"。于是这两个线程会同时执行 [2][3]：

• 假设当前 tail_ = 5，两个线程都读到 tail_ 的值是 5，都把自己的数据写到 buffer_[5]（数据覆盖丢失！）；
• 然后两个线程都各自把 tail_ 更新成 next(5) = 6，而不是期望中的 7（一次更新被"吃掉"了）。
  这就是经典的竞态条件（race condition）——信号量解决的是"有没有资源可用"，但解决不了"多个线程同时修改共享变量"这个问题。这两者是两码事，必须分开处理：数量控制交给信号量，互斥访问交给锁。

4. 第二版：加上互斥锁修复竞态

void push(const T& item)
{
    sem_empty_.acquire();                        // [1] 占用一个空槽位名额（可能阻塞）
    std::unique_lock<std::mutex> lock(mtx_);      // [2] 加锁，保证下面对 buffer_/tail_ 的修改是互斥的
    buffer_[tail_] = item;                        // [3] 写入数据
    tail_ = next(tail_);                           // [4] 更新写指针
    lock.unlock();                                 // [5] 主动解锁（也可以让 lock 出作用域自动解锁）
    sem_full_.release();                           // [6] 释放一个满槽位名额，唤醒可能在等待的消费者
}

对比第一版，唯一的变化是在真正读写 buffer_/tail_ 之前先用 std::unique_lock<std::mutex> 加锁（[2]），操作完成之后再解锁（[5]）。这样即使 sem_empty_.acquire() 同时放行了两个生产者线程，它们对 buffer_ 和 tail_ 的实际修改也会被互斥锁强制串行化，不会再互相踩踏。
一个容易忽略但很重要的细节：信号量的 acquire/release 依然放在锁的外面。这是故意的——acquire() 可能会长时间阻塞（等消费者腾出空位），如果把它放进锁的范围内，会导致这个线程在阻塞等待的同时还占着锁，别的线程完全没法进入临界区，等于把整个队列都锁死了。所以顺序必须是：先用信号量做"资源数量"的把关，再用锁做"互斥访问"的把关。

4.1 try_push：非阻塞版本

bool try_push(const T& item) {
    if (!sem_empty_.try_acquire()) {   // [1] 尝试获取空槽位名额，不阻塞，立即返回成功/失败
        return false;                   // 拿不到令牌（队列可能已满，也可能只是"假失败"）就直接放弃
    }
    std::unique_lock<std::mutex> lock(mtx_);  // [2] 拿到令牌后，加锁保证互斥
    buffer_[tail_] = item;
    tail_ = next(tail_);
    lock.unlock();
    sem_full_.release();
    return true;
}

和 push 相比，唯一区别就是 [1]：用 try_acquire() 代替 acquire()。try_acquire() 不会阻塞，它要么立刻拿到令牌返回 true，要么立刻返回 false。
需要特别注意的一点：try_acquire() 允许**“伪失败”（spurious failure）**——也就是说，即使此刻信号量计数确实大于 0（理论上应该能拿到），try_acquire() 也可能返回 false。这是标准允许的行为（通常是为了在某些平台上换取更好的性能），调用方不能把 try_acquire() 返回 false 简单等同于"资源一定不够用了"。
另外注意注释里提到的一点：try_push 可能仍然会在互斥锁上阻塞——因为一旦通过了信号量这一关，接下来 std::unique_lock<std::mutex> lock(mtx_) 是普通的阻塞加锁，如果此时另一个线程正持有这把锁，当前线程还是要等它释放。所以 try_push 只是"信号量这一步不阻塞"，并不是"整个函数完全不会阻塞"。
try_pop 与 try_push 完全对称，把 sem_empty_/sem_full_ 的角色互换即可，这里不再重复讲解。

5. push / pop 的执行流程时序图

下面用时序图展示两个生产者线程和一个消费者线程之间，是怎么通过 sem_empty_、sem_full_、mtx_ 三者配合完成同步的（假设队列容量为 2，初始为空）：

可以看到：sem_empty_/sem_full_ 负责"排队等资源"，mtx_ 负责"轮到你了也要排队进临界区"，两者配合但各司其职。

6. 完整可运行代码

下面是完整的 semaphore_queue 实现，并附带一个 main 函数：启动 2 个生产者线程和 2 个消费者线程，演示信号量队列在多线程场景下的正确性。代码使用具体头文件而非万能头文件，可以直接用支持 C++20 的编译器编译运行（例如 g++ -std=c++20 -pthread main.cpp -o main）。

// main.cpp
#include <semaphore>   // std::counting_semaphore
#include <mutex>       // std::mutex, std::unique_lock
#include <vector>      // std::vector
#include <thread>      // std::thread
#include <iostream>    // std::cout
#include <atomic>      // std::atomic
#include <cstddef>     // std::size_t
namespace async_prog {
template <typename T>
class semaphore_queue {
public:
    // 构造函数：capacity 是队列容量
    // sem_empty_ 初始化为 capacity（一开始全是空槽位）
    // sem_full_  初始化为 0（一开始没有任何数据）
    explicit semaphore_queue(std::size_t capacity)
        : sem_empty_(static_cast<std::ptrdiff_t>(capacity)),
          sem_full_(0),
          capacity_{capacity},
          buffer_(capacity)
    {}
    // 阻塞式写入：队列满时会一直等待，直到有空位
    void push(const T& item) {
        sem_empty_.acquire();                       // 占用一个空槽位名额（可能阻塞）
        std::unique_lock<std::mutex> lock(mtx_);     // 加锁保护 buffer_/tail_
        buffer_[tail_] = item;                       // 写入数据
        tail_ = next(tail_);                         // 写指针前移（环绕）
        lock.unlock();                               // 提前解锁，缩小临界区
        sem_full_.release();                         // 释放一个满槽位名额，唤醒消费者
    }
    // 非阻塞式写入：队列满时立即返回 false，不等待
    bool try_push(const T& item) {
        if (!sem_empty_.try_acquire()) {             // 尝试获取空槽位名额，不阻塞
            return false;                             // 拿不到就直接放弃（也可能是伪失败）
        }
        std::unique_lock<std::mutex> lock(mtx_);
        buffer_[tail_] = item;
        tail_ = next(tail_);
        lock.unlock();
        sem_full_.release();
        return true;
    }
    // 阻塞式读取：队列空时会一直等待，直到有数据
    void pop(T& item) {
        sem_full_.acquire();                         // 占用一个满槽位名额（可能阻塞）
        std::unique_lock<std::mutex> lock(mtx_);      // 加锁保护 buffer_/head_
        item = buffer_[head_];                        // 读取数据
        head_ = next(head_);                          // 读指针前移（环绕）
        lock.unlock();                                 // 提前解锁
        sem_empty_.release();                          // 释放一个空槽位名额，唤醒生产者
    }
    // 非阻塞式读取：队列空时立即返回 false，不等待
    bool try_pop(T& item) {
        if (!sem_full_.try_acquire()) {
            return false;
        }
        std::unique_lock<std::mutex> lock(mtx_);
        item = buffer_[head_];
        head_ = next(head_);
        lock.unlock();
        sem_empty_.release();
        return true;
    }
private:
    // 计算环绕后的下一个下标：(idx + 1) mod capacity_
    [[nodiscard]] std::size_t next(std::size_t idx) const noexcept {
        return (idx + 1) % capacity_;
    }
private:
    std::mutex mtx_;                        // 保护 buffer_/head_/tail_ 的互斥锁
    std::counting_semaphore<> sem_empty_;   // 空槽位计数信号量
    std::counting_semaphore<> sem_full_;    // 满槽位计数信号量
    std::size_t head_{0};                   // 下一个读取位置
    std::size_t tail_{0};                   // 下一个写入位置
    std::size_t capacity_;                  // 队列容量
    std::vector<T> buffer_;                 // 底层存储
};
} // namespace async_prog
int main() {
    using async_prog::semaphore_queue;
    semaphore_queue<int> q(4);              // 容量为4的信号量队列
    std::atomic<int> produced_sum{0};
    std::atomic<int> consumed_sum{0};
    constexpr int ITEMS_PER_PRODUCER = 50;
    // 两个生产者线程，各自写入 50 个数字
    auto producer = [&](int start_value) {
        for (int i = 0; i < ITEMS_PER_PRODUCER; ++i) {
            int value = start_value + i;
            q.push(value);
            produced_sum.fetch_add(value);
        }
    };
    // 两个消费者线程，各自消费 50 个数字（总数正好和生产者匹配）
    auto consumer = [&]() {
        for (int i = 0; i < ITEMS_PER_PRODUCER; ++i) {
            int value{};
            q.pop(value);
            consumed_sum.fetch_add(value);
        }
    };
    std::thread p1(producer, 0);
    std::thread p2(producer, 1000);
    std::thread c1(consumer);
    std::thread c2(consumer);
    p1.join();
    p2.join();
    c1.join();
    c2.join();
    std::cout << "生产总和: " << produced_sum.load() << '\n';
    std::cout << "消费总和: " << consumed_sum.load() << '\n';
    std::cout << (produced_sum.load() == consumed_sum.load()
                      ? "结果一致，队列同步正确\n"
                      : "结果不一致，说明存在同步问题\n");
    return 0;
}

代码说明：

• semaphore_queue 的构造函数把 sem_empty_ 初始化为 capacity，sem_full_ 初始化为 0，这正好对应第 2 节里讲的"一开始全是空槽位、没有满槽位"。
• main 函数启动了 2 个生产者、2 个消费者线程，一共生产 100 个数、消费 100 个数。因为每个数在写入和读取时都会分别累加到 produced_sum 和 consumed_sum 里（用 std::atomic 保证累加本身的原子性），只要队列的同步逻辑正确，两个总和必然相等——这就是一个简单但有效的正确性验证。
• 如果把 push/pop 里的 std::unique_lock<std::mutex> lock(mtx_) 去掉（也就是退化回第 3 节"有漏洞的版本"），在多生产者/多消费者场景下多次运行这个程序，会大概率出现"生产总和 ≠ 消费总和"的情况，这就直观验证了为什么必须加锁。

7. 小结

• 计数信号量本质是一个"计数令牌桶"：acquire() 拿令牌（不够就等待），release() 还令牌（同时可能唤醒等待者）。
• 信号量非常适合用来表示"某种资源还剩多少个"，比如环形队列里的"空槽位数"和"满槽位数"。
• 信号量只管数量，不管互斥：多个线程可能同时拿到令牌，然后同时闯入对共享数据的读写，所以数量控制（信号量）和互斥访问（std::mutex）要配合使用，缺一不可。
• try_acquire() 提供了非阻塞的调用方式，方便线程在暂时拿不到资源时去做别的事情；但要留意它允许"伪失败"，也要留意 try_push/try_pop 里紧跟着的互斥锁本身仍然可能阻塞。
• 计数信号量（std::counting_semaphore）是 C++20 新加入标准库的同步原语，和条件变量（std::condition_variable）相比，写生产者-消费者模型时代码通常更直接、更不容易遗漏"信号丢失"之类的坑。

二进制信号量（Binary Semaphore）从零理解

1. 什么是信号量

信号量本质上是一个"计数器 + 等待队列"的组合体。可以把它想象成一个仓库门口挂着的一块牌子，牌子上写着一个数字，这个数字表示"还能有多少人进去"。

• 当有人想进去（这在代码里叫 acquire / wait），就先看牌子上的数字：
  • 如果数字大于 0，就把数字减一，然后自己进去；
  • 如果数字等于 0，说明"没名额了"，这个人就必须在门口等着，直到有人出来腾出名额。
• 当有人出来（这在代码里叫 release / signal），就把牌子上的数字加一，如果门口正好有人在等，就会唤醒其中一个人进去。
  用数学语言描述，设信号量内部计数器为 $S$，则：
  $$\text{acquire}(S):\text{若 }S>0,S\leftarrow S-1;\text{否则阻塞等待，直到 }S>0$$
  $$\text{release}(S):S\leftarrow S+1,\text{并唤醒一个等待中的线程（如果有）}$$
  二进制信号量是信号量的一种特殊情况：它的计数器只能取 $0$ 或 $1$ 这两个值（所以叫"二进制"）。也就是说 S∈{0,1}S \in \{0, 1\}S∈{0,1}。

---

信号量值	含义
1	资源"可用"，此时 acquire 不会阻塞
0	资源"不可用"，此时 acquire 会阻塞，直到别的线程 release

---

2. 二进制信号量 vs 互斥锁（mutex）：区别在哪

很多初学者会把二进制信号量和互斥锁搞混，因为表面上看它们都是"要么锁上、要么打开"的两态开关。但它们的语义（谁能做什么操作）完全不同。

对比维度	互斥锁 Mutex	二进制信号量 Binary Semaphore
所有权	有"归属"，只有加锁的那个线程才能解锁	没有归属，任何线程都能把它 release
用途定位	保护临界区，实现互斥访问	更像一种"信号通知"机制
类比	“我锁上的门，只有我能开”	“谁看到信号灯变绿都可以按一下重置它”
更接近谁	自己（独占的锁）	条件变量（condition variable）

---

也正因为二进制信号量"谁都能 release"这个特点，它天然适合用来做线程之间的信号通知（比如：线程 A 做完一件事，"叫醒"线程 B 开始做下一件事），而不是简单的互斥锁替代品。

3. C++20 中的 std::binary_semaphore

C++20 在标准库中引入了 <semaphore> 头文件，提供了模板类 std::counting_semaphore<LeastMaxValue>。
std::binary_semaphore 只是它的一个"特化别名"，等价于把最大值 LeastMaxValue 固定为 1：

using binary_semaphore = std::counting_semaphore<1>;

初始化时必须显式给出初始值，只能是 0 或 1：

std::binary_semaphore sm1{ 0 };  // 初始"不可用"，第一个 acquire 会被阻塞
std::binary_semaphore sm2{ 1 };  // 初始"可用"，第一个 acquire 立刻成功

对应的核心操作（成员函数）：

操作	作用	等价理解
acquire()	计数器减一，若已是 0 则阻塞等待	上面说的"进门"
release()	计数器加一，并唤醒一个等待者	上面说的"出门，让下一个人进"
try_acquire()	非阻塞尝试获取，成功返回 true，失败立刻返回 false	“看一眼，不排队”

---

4. 代码示例一：用二进制信号量实现"信号通知"（推荐用法）

下面这个例子模拟"线程 A 准备好数据后，通知线程 B 去处理数据"的场景。这正好体现了二进制信号量作为信号机制而非互斥锁的本质用法。

#include <iostream>   // 用于标准输入输出 std::cout
#include <semaphore>  // C++20 提供的信号量头文件，包含 std::binary_semaphore
#include <thread>     // 用于创建和管理线程 std::thread
#include <chrono>     // 用于线程休眠，模拟耗时操作
// 初始值为 0，表示"数据还没准备好"，
// 因此消费者线程一开始 acquire() 时会被阻塞，必须等待生产者 release()。
std::binary_semaphore data_ready{ 0 };
int shared_data = 0; // 生产者和消费者共同访问的"共享资源"
// 生产者线程：负责生产数据，然后发出"数据已就绪"的信号
void producer()
{
    std::cout << "[生产者] 正在生成数据...\n";
    std::this_thread::sleep_for(std::chrono::seconds(1)); // 模拟耗时的准备工作
    shared_data = 42; // 写入共享数据
    std::cout << "[生产者] 数据已准备好，发出信号！\n";
    data_ready.release(); // 计数器从 0 变为 1，如果消费者正在等待，会被唤醒
}
// 消费者线程：等待信号，信号到达后才读取数据
void consumer()
{
    std::cout << "[消费者] 等待数据信号...\n";
    data_ready.acquire(); // 计数器是 0，这里会阻塞，直到生产者 release()
    std::cout << "[消费者] 收到信号，读取数据：" << shared_data << "\n";
}
int main()
{
    // 分别启动生产者线程和消费者线程
    std::thread t1(producer);
    std::thread t2(consumer);
    // 等待两个线程都执行完毕，防止 main 提前退出
    t1.join();
    t2.join();
    return 0;
}

代码逐点解析

1. std::binary_semaphore data_ready{ 0 };
   把信号量的初始值设为 0，意思是"目前没有可用信号"，任何线程一上来 acquire() 都会被卡住。这一步是整个例子的关键设计：我们故意让消费者"先天缺信号"，逼它必须等生产者。
2. producer() 函数里：
   • 先用 sleep_for 模拟一段"制造数据"需要花费的时间；
   • 把 shared_data 赋值为 42，这是真正的共享资源写入操作；
   • 最后调用 data_ready.release()，把计数器从 0 加到 1，这一步如果此时消费者正卡在 acquire() 里，就会立刻把它唤醒。
3. consumer() 函数里：
   • 调用 data_ready.acquire()，因为初始值是 0，这里线程会阻塞在这一行，什么都不做，直到别的线程调用了 release()；
   • 一旦被唤醒（计数器被减回 0），才会执行下一行，把 shared_data 打印出来。
4. main() 里同时启动两个线程 t1（生产者）和 t2（消费者），谁先被操作系统调度执行都没关系——即使消费者先跑，它也只会卡在 acquire() 处干等，绝不会提前读到脏数据，这正是信号量"排序保证"的价值所在。

时序图（说明信号如何在两个线程间传递）

5. 代码示例二：用二进制信号量实现互斥（了解即可，不推荐）

虽然前面强调"信号量不适合做互斥"，但为了理解它和 mutex 的差异，这里也演示一下：如果硬要用二进制信号量当锁来用，代码长什么样。

#include <iostream>
#include <semaphore>
#include <thread>
#include <vector>
// 初始值为 1，表示"资源可用"，这一点和 mutex 的"未加锁"状态类似
std::binary_semaphore resource_lock{ 1 };
int counter = 0; // 被多个线程共享并竞争修改的变量
void worker(int id)
{
    for (int i = 0; i < 100000; ++i)
    {
        resource_lock.acquire(); // 相当于"加锁"：计数器由1变0
        ++counter;               // 临界区：安全地修改共享变量
        resource_lock.release(); // 相当于"解锁"：计数器由0变1
        // 注意：这里 release() 可以被任何线程调用，
        // 即便不是当初 acquire 的那个线程也一样能调用成功，
        // 这就是它和 mutex "谁加锁谁解锁"的本质区别。
    }
    std::cout << "[线程 " << id << "] 完成\n";
}
int main()
{
    std::vector<std::thread> threads;
    for (int i = 0; i < 4; ++i)
        threads.emplace_back(worker, i); // 启动4个线程，一起竞争 counter
    for (auto& t : threads)
        t.join();
    std::cout << "最终 counter 的值 = " << counter << "\n"; // 应为 400000
    return 0;
}

代码逐点解析

1. std::binary_semaphore resource_lock{ 1 }; 初值设为 1，表示一开始资源是"空闲可用"的，第一个线程调用 acquire() 不会被阻塞。
2. 每个 worker 线程循环十万次，每次循环都：
   • acquire()：尝试把计数器减到 0，如果已经有别的线程占用（计数器是 0），这里就会阻塞排队；
   • ++counter：这是被保护的"临界区"代码，因为同一时刻只有一个线程能持有计数器为 0 的状态，所以这行代码不会有数据竞争；
   • release()：把计数器加回 1，让下一个排队的线程能进入。
3. 关键风险点：release() 不检查调用者是不是当初 acquire 的那个线程。这意味着如果代码写错了（比如线程 A 忘记 acquire，却手滑调用了 release），计数器可能被错误地加到超过预期状态，导致多个线程同时进入临界区，产生数据竞争。这正是"没有所有权"带来的隐患，也是为什么真正做互斥时官方推荐用 std::mutex 而不是二进制信号量。

6. 状态转换图（ASCII 版）

用 ASCII 图直观展示二进制信号量的两态转换：

                acquire()（计数器 1 -> 0）
        +------------------------------------+
        |                                    |
        v                                    |
  +-----------+                        +-----------+
  |  值 = 1   |                        |  值 = 0   |
  | (可用/开)  |                        | (占用/关)  |
  +-----------+                        +-----------+
        ^                                    |
        |                                    |
        +------------------------------------+
                release()（计数器 0 -> 1）
  注：若某线程在"值=0"时调用 acquire()，
      该线程会被阻塞，直到别的线程调用 release() 使值变回 1。

7. 小结

• 二进制信号量的计数器只能是 $0$ 或 $1$，本质是"资源可用/不可用"的开关。
• 它和互斥锁最大的不同在于所有权：mutex 只有加锁者能解锁；二进制信号量任何线程都能 release，因此更接近条件变量，天然适合做跨线程信号通知（如例子一：生产者通知消费者），而不建议单纯拿来做互斥（如例子二，虽然能凑效但存在被误用的风险）。
• C++20 通过 std::binary_semaphore（即 std::counting_semaphore<1> 的别名）提供了标准库支持，核心接口是 acquire() / release() / try_acquire()。
  编译运行示例代码（Linux/g++，需要 C++20 及线程库支持）：

g++ -std=c++20 -pthread example.cpp -o example
./example