Linux 线程同步与互斥：原子性、互斥锁、死锁与条件变量

上篇文章我们亲手创建了线程，也见识了线程之间靠"共享内存"互相交流的威力。但威力往往伴随着危险——当多个线程同时去读写同一个变量时，结果可能会出乎你的意料。

这篇文章是整个线程系列里最考验"火候"的一篇：我们要讲清楚三件事。第一，为什么"多个线程操作同一个变量"会出错，以及"原子性"到底是什么意思；第二，如何用 Linux 提供的互斥量（mutex）让访问临界资源的代码变得安全；第三，如何用条件变量（cond）实现线程与线程之间的"同步"——也就是让线程按我们希望的顺序干活，而不是瞎抢。

概念很多，代码会很长，但我会给每一行都配上注释，并保证每个例子都能在 Linux 下编译运行。你只需要有一条能够 g++ 或者 gcc 的环境，编译时记得加 -pthread。我们开始。

为什么多线程卖票会卖亏本

先看一道非常经典的教学题目：模拟一个售票系统，全场一共 100 张票，开 4 个线程同时去卖。直观想，票必须是"卖一张少一张"，最终 4 个线程一共恰好卖出 100 张，一张不多一张不少。但如果你不加任何保护，直接这样写，结果会让你大跌眼镜。

// sell_bug.c   ——  不加任何保护的售票系统
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>    // usleep 所在，用来模拟"卖一张票要花一点时间"
#include <pthread.h>   // 线程相关接口
 
int ticket = 100;      // 共享变量：全场总票数。4 个线程都能看到它
 
void *route(void *arg) // 每个卖票线程的入口函数
{
    char *id = (char *)arg;          // 拿到这个线程的名字，用于打印
    while (1)                        // 死循环抢票
    {
        if (ticket > 0)              // 还有票？
        {
            usleep(1000);            // 模拟"漫长的售票业务"，耗 1 毫秒
            printf("%s sells ticket:%d\n", id, ticket); // 打印当前票号
            ticket--;                // 卖掉一张，票数减一
        }
        else                         // 没票了
        {
            break;                   // 退出循环，线程结束
        }
    }
    return NULL;
}
 
int main(void)
{
    pthread_t t1, t2, t3, t4;        // 4 个线程的 id
    // 依次创建 4 个卖票线程，参数是给线程起的名字
    pthread_create(&t1, NULL, route, (void *)"thread 1");
    pthread_create(&t2, NULL, route, (void *)"thread 2");
    pthread_create(&t3, NULL, route, (void *)"thread 3");
    pthread_create(&t4, NULL, route, (void *)"thread 4");
    // 等 4 个线程都跑完
    pthread_join(t1, NULL);
    pthread_join(t2, NULL);
    pthread_join(t3, NULL);
    pthread_join(t4, NULL);
    return 0;                        // 主线程正常结束
}

编译运行：

gcc sell_bug.c -o sell_bug -pthread   # -pthread 必须加，否则链接不到线程库
./sell_bug

一次真实的运行结果（每次运行输出都可能不一样，但往往都不对）：

thread 4 sells ticket:100
...
thread 4 sells ticket:1
thread 2 sells ticket:0
thread 1 sells ticket:-1
thread 3 sells ticket:-2

看到了吗？不仅出现了 ticket:0，还出现了 ticket:-1、ticket:-2。票不但被卖光了，还被"卖穿了"——卖出了负数票号的票。这明显是逻辑错误：明明 if (ticket > 0) 检查过还有票，为什么最后还能把票卖到 -2？

这就是不保护共享变量的后果。要真正想明白它，我们必须把"原因"拆成三层来理解，缺一不可。

原因一：判断"还有票"之后，线程可能被切换走

if (ticket > 0) 这个判断本身只是一瞬间的事。但关键在于：线程的运行是"分时复用"的，操作系统随时可能把一个正在运行的线程切出去，换另一个线程跑（这叫抢占式调度）。

设想这个时序：

1. 线程 A 执行 if (ticket > 0)，看到 ticket = 5，条件成立，正准备卖第 5 张票。
2. 就在这一刻，A 被操作系统切出去，让出 CPU。
3. 线程 B 也执行 if (ticket > 0)，看到 ticket = 5，也认为还有票，于是 B 卖出第 5 张，ticket 减成 4。
4. 线程 C 又跑进来，看到 ticket = 4，也认为是"还有票"，卖出第 4 张，ticket 减成 3……
5. 一批线程抢着把票从 5 卖到了 0、-1、-2。
6. 这时候 A 才重新拿到 CPU，继续执行它"卖出第 5 张票"的余下代码，卖出 5、甚至 4、3……都是早已被卖过的票号；再加 . 上 ticket-- 一堆线程同时减，负数就出来了。

你可能会质疑：if (ticket > 0) 和 ticket-- 之间不就隔了一行 usleep(1000) 吗？恰恰是这行 usleep 制造了"漫长的业务过程"，在执行它的这一毫秒里，足够其他线程涌进来把票抢光、抢穿。这就是为什么代码里要特意放一个 usleep——它把"其实会发生"的切换放大了，让你能肉眼看到错误。

原因二：ticket-- 根本不是一个原子操作

这是最关键、也最反直觉的一点：写成一条语句的 ticket--，在底层其实是三条机器指令。我们可以用 objdump 把它反汇编出来看看，到底是哪三条。

objdump -d sell_bug > test.objdump     # 把可执行文件反汇编，重定向到文件

在 test.objdump 里找到 ticket-- 对应的汇编，大致长这样（不同平台寄存器名略有差异，但结构一定是这三步）：

mov    0x2004e3(%rip),%eax    # 指令1 load :  把内存里的 ticket 读到寄存器 eax
sub    $0x1,%eax              # 指令2 update: 在寄存器里做减一
mov    %eax,0x2004da(%rip)    # 指令3 store : 把寄存器里新的值写回内存里的 ticket

翻译成人话：

• load：把共享变量 ticket 从内存加载到（某个线程自己私有的）寄存器里；
• update：在寄存器里 -1，得到新值；
• store：把新值从寄存器写回 ticket 所在的内存地址。

现在设想两个线程同时拿到同一时刻的 ticket = 5，然后各自执行这三条指令，就可能出现这样的交错：

1. 线程 A load：A 的寄存器 = 5；
2. 线程 B load：B 的寄存器 = 5（此时 ticket 内存里还是 5）；
3. 线程 A update：A 的寄存器 = 4；
4. 线程 A store：ticket 内存 = 4；
5. 线程 B update：B 的寄存器 = 4；
6. 线程 B store：ticket 内存 = 4。

原本应该"卖两张票"，结果 ticket 只从 5 减到了 4——丢失了一次更新。这就是经典的"读-改-写三态不等变"问题。++ 和 i++ 也一样会踩中：它们都不是原子的。

原因三：对共享变量的访问互相踩踏

上面两条已经足够说明问题了，但把它们放一起，本质是同一个词：竞态。多个线程"不设防"地访问同一个共享变量，谁先谁后完全不可控，结果取决于最后一次"谁先 store"，从而产生雪花一样不确定的错误。我们把这种由时序（谁先谁后）导致程序异常的现象，统称为竞态条件。竞态是本篇文章所有问题的总根源。

思考题 1：为什么 "ticket > 0 判断为真" 并不能保证 "ticket 一定 > 0"？请结合"判断"与"减一"两条语句之间的切换，给出完整解释。

答案

"判断 ticket > 0"和"ticket--"是两条独立的语句，它们之间没有任何原子性保证。当线程 A 判断此刻 ticket>0 为真后，它在执行 usleep 的过程中被 OS 切换走，其他线程可以一拥而上把 ticket 卖到负数。等到 A 重新被调度回，它直接执行"卖出这张票"，但此时 ticket 早已不是当初它看到的那个正数。也就是说：判断"发生的那一刻"票有货，但"真正卖票的那一刻"票可能已经没了。中间的这段窗口期里对 ticket 的修改，A 完全感知不到。因此判断为真 ≠ 卖票时仍有票。

关键概念：共享资源、临界区、互斥与原子性

在动手写锁之前，先把四个名词一次性讲清楚。它们是整个同步领域的"地基"。

• 共享资源：能被多个执行流（线程、进程）共同访问的资源。例如上面的全局变量 ticket、后面会出现的队列、显示器、文件等。注意"能被共享"不等于"就该不分时共享"。
• 临界资源：需要被"特殊保护"才能安全使用的共享资源。换句话说，临界资源 = 共享资源 + 保护需求。多数共享资源在没有保护的情况下被并发访问就会坏掉，于是我们在谈论"要被保护起来的共享资源"时，就叫它临界资源。
• 临界区：线程内部，真正去访问临界资源的那段代码。同一个全局变量，可能被 4 个线程各自访问，那么这 4 段"访问它的代码"就都是临界区。需要强调：临界区是"代码段"的概念，不是"数据"的概念。
• 互斥：保证"任何时刻，有且只有一条执行流能进入临界区"的一种机制。互斥的作用是给临界资源"上锁"，把多条同时要访问的流，强制串行化成"一条一条进"。
• 原子性：一个操作如果"不会被任何调度机制打断"，我们就说它具有原子性。原子操作只有两种结果：要么整个完成，要么整个未完成，绝不出现"做了一半"。正因为 ticket-- 不是原子操作，才会出现前面那个坏结果。

课件里给原子性下的定义很精准：不会被任何调度机制打断的操作，该操作只有两态：要么完成，要么未完成。什么叫"被打断"？就是操作系统在它执行的中途把线程切走了。三条汇编才能做完的 --，就可能在第一条和第二条之间被切断，于是"做了一半"——原子性被破坏。

思考题 2：为什么说"原子操作只有两种结果，要么完成要么未完成"？这和"中间态"有什么关系？

答案

非原子操作的执行会被拆成若干步，执行到任意两步之间都可能被切换，于是外部观察者可能看到"它已经改了内存但它又没完全改完"的中间状态。原子操作由于不会被切断，外部永远看不到它"做了一半"——要么它在切换发生前整个做完（新值已落内存），要么它还没开始（旧值仍在内存），不存在"内存里是混乱的中间值"这种情况。这个"没有中间态"的特性，正是把共享数据做一致性判断时最需要的东西。

互斥量 mutex 是什么

要让临界区"有且只能有一个线程进"，本质上我们需要一把"锁"：一个人拿走了，别人就进不来；那人用完还回来，下一个人才能进。Linux / POSIX 提供的这把标准的锁，就叫互斥量（Mutual Exclusion，缩写 mutex）。

互斥量天然满足开头我们列的那三点要求：

1. 当某个线程进入了临界区（拿到了锁），其他线程不允许再进入（拿不到锁）；
2. 如果多个线程同时要进临界区、而临界区此刻没人，那么只有其中一个能抢到锁，其他都得等；
3. 不在临界区里的线程，不会妨碍其他线程抢锁。

小科普：互斥量的"量"字，呼应的是"信号量"里的量。互斥量可看作"取值只能是 0 或 1 的一种特殊信号量"——0 表示锁被占，1 表示锁空闲。不过它和信号量有两处关键不同，我们讲信号量时会再对比，先记住"互斥量 ≈ 二值锁"这个直觉即可。

互斥量的接口一览

互斥量在头文件 <pthread.h> 中，类型是 pthread_mutex_t。围绕它有一组配套函数：

需要 -pthread：因为它们都在线程库里，编译时若不加 -pthread，链接阶段会报"找不到 pthread_create"这类错误。

怎么初始化互斥量

初始化有两种方法，两者选其一即可，不要混用。

方法一：静态分配（最简单）。 用宏 PTHREAD_MUTEX_INITIALIZER 直接在定义时就初始化好：

pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;   // 定义的同时完成初始化

这种方法适合"全局互斥量、在程序启动前就需要就绪"的场景，因为它不需要调用任何函数，也不需要在用之前单独 init。

方法二：动态分配。 调用 pthread_mutex_init，在运行时初始化：

#include <pthread.h>
 
int pthread_mutex_init(pthread_mutex_t *restrict mutex,
                       const pthread_mutexattr_t *restrict attr);

参数：

• mutex：要初始化的互斥量指针；
• attr：互斥量的属性。这一节我们一律传 NULL，表示用默认属性（普通、非递归、默认互斥）。等以后学"递归锁/读写锁"时才需要用属性去定制；
• 返回值：成功返回 0；失败返回错误码（注意，POSIX 线程函数失败时返回错误号，而不是像 errno 那样的全局变量，也不返回 -1）。

pthread_mutex_t mutex;      // 先声明一个互斥量
pthread_mutex_init(&mutex, NULL);   // 运行时用默认属性初始化它

两种方法不要同时用——对同一个互斥量先 PTHREAD_MUTEX_INITIALIZER 又 pthread_mutex_init，属于未定义行为。

怎么销毁互斥量

用 pthread_mutex_destroy：

int pthread_mutex_destroy(pthread_mutex_t *mutex);

销毁有三个"雷区"，课件里反复强调，我补充完整：

1. 用 PTHREAD_MUTEX_INITIALIZER 静态初始化的互斥量，不需要（也不应该）销毁——它随程序的启动而就绪、随程序结束而释放，没有动态资源需要回收。
2. 绝不能销毁一个正在被某个线程锁着的互斥量。销毁一个"锁着"的锁，结果是未定义的，后面等着用它加锁的线程会彻底乱套。
3. 销毁之后，要确保没有其他线程还要对这把已经销毁的锁加锁。销毁不代表"别的线程不认识它了"，一旦有线程再去 lock 这把已经 destroy 的锁，就是未定义行为。

一句话总结：锁的生命周期管理要非常克制——"在哪初始化，就在哪销毁；先解锁，再销毁；销毁后别再碰"。

加锁与解锁的行为

int pthread_mutex_lock(pthread_mutex_t *mutex);   // 加锁
int pthread_mutex_unlock(pthread_mutex_t *mutex); // 解锁，成功返回 0，失败返回错误号

pthread_mutex_lock 调用时会出现两种情况：

• 互斥量此刻处于未锁状态：函数把锁置为已锁，然后返回成功，调用者进入临界区；
• 互斥量此刻已被别的线程锁定，或者多个线程同时申请但你没抢到：调用会阻塞（执行流被挂起），直到持有锁的线程 unlock 后，锁被让出来，才继续返回。这一层"抢不到就睡、锁释放就醒"的调度，全部由内核/线程库帮我们完成，程序员看不到。

至于 pthread_mutex_trylock，它和 lock 的唯一区别是：拿不到锁时不阻塞，立即返回一个错误号（EBUSY）。它适合"不想空等、先干点别的事再来看锁"的场景。

unlock 的规则是：谁加的锁，谁来解。一个线程加的锁，只能由它自己解锁。让 A 锁、B 解，属于未定义行为，是初学最容易埋的雷之一。

改进：给售票系统加上互斥量

现在我们把锁用起来。THINK 一下应该在哪些地方加锁、哪些地方解锁：凡是"读 ticket 判断、卖票、ticket--"这一连串动作，都必须放在同一把锁 → 同一个临界区里，因为它们是一个整体业务，不允许被拆开。特别是 if (ticket > 0) 和 ticket-- 必须一起拿锁。

// sell_lock.c   ——  用互斥量保护售票系统
#include <stdio.h>
#include <string.h>
#include <unistd.h>
#include <pthread.h>
 
int ticket = 100;                    // 共享变量：总票数
pthread_mutex_t mutex;               // 定义一把互斥量（稍后用 init 动态初始化）
 
void *route(void *arg)
{
    char *id = (char *)arg;
    while (1)
    {
        pthread_mutex_lock(&mutex);      // 进入临界区前加锁：拿不到就一直等
        if (ticket > 0)                  // 判断还有票（此刻独占访问，安全）
        {
            usleep(1000);                // 模拟漫长的售票业务
            printf("%s sells ticket:%d\n", id, ticket); // 打印票号
            ticket--;                    // 减一（此刻独占访问，安全）
            pthread_mutex_unlock(&mutex);// 卖完立刻解锁，放别人进来
        }
        else                             // 没有票了
        {
            pthread_mutex_unlock(&mutex);// 即使要退出，也必须先解锁！
            break;                       // 退出循环
        }
    }
    return NULL;
}
 
int main(void)
{
    pthread_t t1, t2, t3, t4;
    pthread_mutex_init(&mutex, NULL);    // 动态初始化互斥量（默认属性）
    pthread_create(&t1, NULL, route, (void *)"thread 1");
    pthread_create(&t2, NULL, route, (void *)"thread 2");
    pthread_create(&t3, NULL, route, (void *)"thread 3");
    pthread_create(&t4, NULL, route, (void *)"thread 4");
    pthread_join(t1, NULL);
    pthread_join(t2, NULL);
    pthread_join(t3, NULL);
    pthread_join(t4, NULL);
    pthread_mutex_destroy(&mutex);       // 所有线程结束后再销毁锁
    return 0;
}

一个非常容易忽略、却极其重要的细节，请盯着看：在 else（没票）分支里，我也调用了 pthread_mutex_unlock 才 break。为什么？

因为 break 只是"跳出了 while 循环"，它不会主动帮我们还锁。如果我们在"还有票"的逻辑里加了锁、在"没票"的路径上忘了解锁就直接 break，那么这把锁就会永远被这个即将退出的线程霸占，其他三个线程会永远阻塞在 pthread_mutex_lock 上，程序变成一个"假死"状态。"加锁的路径有多少个出口，解锁就得覆盖多少个出口"——这是写锁代码最容易踩的坑，后文讲 RAII 时我们会有更优雅的解法。

编译运行，这次结果正确了：

gcc sell_lock.c -o sell_lock -pthread
./sell_lock

现在票号是严格从 100 一路减到 1，从未出现 0、-1、-2。4 个线程被锁串行化了：同一时刻只有一个线程在"看票 + 卖票"，卖完一把，别人才能接手。

思考题 3：如果把 if (ticket > 0) 的判断放在锁外面、只有 ticket-- 放在锁里面，程序还是错的。请问为什么？

答案

因为"判断还有没有票"和"真正减一卖票"是两段必须绑定在一起的整体动作。如果判断在锁外，那么两个线程可能都在锁外都看到"还有票"，都在锁外都通过了 if，然后进入锁内排队卖票——可实际上票可能只有一个、甚至已经减成负数了，第二个线程还会照卖不误。加锁的粒度必须覆盖"从判断资源是否可用，到真正使用/减少资源"的整段代码，二者之间不能留缝隙。锁只保护它包住的那段代码，代码外面的判断不受保护，是这种写法仍然出错的根本原因。

互斥量的底层原理：锁是怎么做到"原子"的

看到这你可能想问：pthread_mutex_lock 自己加锁的动作，难道不会被切换打断吗？如果"加锁"本身也像 ticket-- 一样不是原子的，那这锁不是形同虚设？

问得好。答案的关键在于：互斥量的加锁动作，底层用了 CPU 提供的"原子交换"指令，比如 x86 的 xchg / cmpxchg，或者"比较并交换"（Compare And Swap，缩写为 CAS）这类指令。

回忆开头的三态问题：ticket-- 之所以坏，是因为 load / update / store 被拆成三条指令，中间能插进别的线程。而原子交换指令则把"检查当前值 + 写入新值"这整个过程压缩成一条机器指令，这条指令在执行期间：

• 单核平台：一条指令不会被中断的粒度破坏（指令执行是连续的）；
• 多核/多处理器平台：访问内存有"总线周期"的先后顺序，一个处理器执行原子交换指令时，会锁住内存总线（或者说用缓存一致性协议在 LOCK 前缀下独占），另一个处理器想同时做原子交换，只能等待这个总线周期先结束。

所以"锁自己"这个动作，天然是原子的。下面我们把 mutex 的 lock / unlock 用"类似 C"的伪代码写出来，体会一下它到底做了什么。假设锁用一个整型 lock = {0} 表示，0 表示没锁，1 表示已锁：

// 伪代码：体会"原子交换"实现一把锁的骨架。真正的 glibc/NPTL 实现远比这复杂，
// 但核心思想是一致的：借助一条原子的交换指令，避免"检查-置位"两态被拆开。
int lock = 0;                          // 全局共享：0=空闲，1=被锁定
 
void my_mutex_lock(volatile int *lock)
{
    // 用一个临时变量 tmp 的初值 1，与内存中的 lock 做原子的交换：
    //   交换后：lock 变成 1（表示服务器已把这个锁标记为"被我占"）
    //           tmp 拿到 lock 原来的值（0 或 1）
    int tmp = 1;
    while (exchange(lock, &tmp))        // 原子交换；tmp 里若还是 1，说明别人持有 → 循环重试
        ;                               // 锁被占，继续自旋（或让出 CPU 挂起）
}
 
void my_mutex_unlock(volatile int *lock)
{
    *lock = 0;                          // 把锁置回 0，归还资源；别的线程 swap 就能拿到
}

这段伪代码里最关键的一行是 while (exchange(lock, &tmp))。exchange 一步完成"把 lock 的旧值拿进 tmp，同时把 tmp 的当前值（1）写入 lock"，而且这一来一回是原子的。为什么要"交换"而不是朴素的"先检查再赋值"？

设想如果不用交换，而是朴素写法——"先 if (lock == 0)，然后 lock = 1"，这两步之间如果有两个线程同时看到 lock==0 并同时把 lock 置 1，锁就破防了，两个人同时进入临界区。而原子交换把"{ 读旧值 ； 写新值 }"绑定成一条指令，永远不可能有两个人同时成功：因为当一个人成功把 lock 置 1 时，另一个人拿到的旧值一定已经是 1，从而 while 判定为"已锁"，乖乖去等。

这也就回应了开头课件的这句提纲："一条 swap/exchange 指令就够了，恰好是因为它只有一条指令、没法被打断，从而保证了原子性"。

思考题 4：有人说"互斥锁加锁时就该一直空转直到拿到锁"，这种写法（自旋）在什么情况下是浪费，什么情况下反而更快？

答案

空转等待叫"自旋锁"（spinlock）。自旋好处是不切换线程、不陷入内核，如果临界区极短、锁马上就会被释放，自旋可以在几纳秒内原地拿到锁，比"睡下再唤醒"廉价得多；坏处是它消耗 CPU——如果临界区较长且持锁线程很少让位，自旋线程会白白烧掉一个核心的算力。PTHREAD_MUTEX 默认不是纯自旋，而是"先自旋一小段时间，抢不到再挂起"，兼顾两者。经验法则：临界区非常短（几十条指令内）时用自旋划算，临界区可能长时间占用时用会"挂起"的互斥量划算。这个 trade-off 我们放到"其他锁"一节还会展开。

用 RAII 风格封装互斥量

前面手工 lock/unlock 的写法有个天然缺陷：解锁很容易漏。一旦函数里出现 return、异常、多出口，就可能在某个出口忘了 unlock，锁被永久霸占。C++ 世界给这种问题提供了一个非常优雅的标准解法——RAII。

RAII（Resource Acquisition Is Initialization），直译"资源获取即初始化"，核心思想是：让资源的"获取"绑定在对象的构造函数里，让资源的"释放"绑定在对象的析构函数里。因为对象的析构函数在对象生命周期结束时（离开作用域）一定会被自动调用，所以只要把"解锁"放进析构函数，就再也忘不掉了——不管你是正常走到底、中间 return、还是抛异常跳出，析构都会执行，锁都会被释放。

课件先提供了一个朴素的封装（Mutex + LockGuard），我们把它完整写出并逐行注释。文件名 Lock.hpp：

// Lock.hpp  ——  将 pthread 互斥量封装成 RAII 风格，方便后续复用
#pragma once                              // 防止头文件被重复包含
#include <pthread.h>                       // pthread 互斥量接口
 
namespace LockModule                       // 自己造的命名空间，避免污染全局
{
    // ---------- Mutex：对 pthread_mutex_t 的薄封装，可独立使用 ----------
    class Mutex
    {
    public:
        // 删除拷贝构造和赋值：锁是"不可复制"的，复制一把锁毫无意义且危险
        Mutex(const Mutex &) = delete;
        const Mutex &operator=(const Mutex &) = delete;
 
        Mutex()                            // 构造：初始化锁
        {
            int n = pthread_mutex_init(&_mutex, nullptr); // 默认属性初始化
            (void)n;                       // 暂时忽略返回值（实际工程应检查并记日志）
        }
 
        void Lock()                        // 加锁
        {
            int n = pthread_mutex_lock(&_mutex);
            (void)n;
        }
 
        void Unlock()                      // 解锁
        {
            int n = pthread_mutex_unlock(&_mutex);
            (void)n;
        }
 
        pthread_mutex_t *GetMutexOriginal() // 暴露原始指针，供条件变量等底层接口使用
        {
            return &_mutex;
        }
 
        ~Mutex()                           // 析构：销毁锁
        {
            int n = pthread_mutex_destroy(&_mutex);
            (void)n;
        }
 
    private:
        pthread_mutex_t _mutex;            // 真正持有的 pthread 互斥量
    };
 
    // ---------- LockGuard：RAII 锁管理器，用作用域自动锁定/自动解锁 ----------
    class LockGuard
    {
    public:
        LockGuard(Mutex &mutex)            // 构造：持有一把锁的引用，并立即上锁
            : _mutex(mutex)
        {
            _mutex.Lock();                 // 进入作用域即加锁
        }
 
        ~LockGuard()                       // 析构：离开作用域自动解锁
        {
            _mutex.Unlock();               // 无论怎么离开，这里一定会执行
        }
 
    private:
        Mutex &_mutex;                     // 持引用，不拷贝锁本身
    };
}

LockGuard 用起来极其省心：你只要在栈上构造一个 LockGuard 对象，它就在构造时上锁、在离开作用域时自动解锁。我们用它重写抢票逻辑：

// sell_guard.cpp   ——  用 RAII 风格的锁重写售票系统
#include <stdio.h>
#include <unistd.h>
#include <pthread.h>
#include "Lock.hpp"                        // 引入我们刚刚封装的互斥锁
 
using namespace LockModule;                // 使用 LockModule 里的类型
 
int ticket = 1000;                         // 共享变量：总票数（改成 1000 增加测试量）
Mutex mutex;                               // 全局一把自己封装的锁
 
void *route(void *arg)
{
    char *id = (char *)arg;
    while (1)
    {
        LockGuard lockguard(mutex);        // 进入 while 这一轮就自动上锁！
        if (ticket > 0)
        {
            usleep(1000);
            printf("%s sells ticket:%d\n", id, ticket);
            ticket--;
            // 这里不需要手动 unlock！
            // 当这一轮 while 的代码块结束、或者 break 跳出时，
            // lockguard 离开作用域，析构函数自动调用 Unlock()，锁必然被释放。
        }
        else
        {
            break;                          // break 也会触发 lockguard 析构，锁照样释放
        }
    }
    return NULL;
}
 
int main(void)
{
    pthread_t t1, t2, t3, t4;
    pthread_create(&t1, NULL, route, (void *)"thread 1");
    pthread_create(&t2, NULL, route, (void *)"thread 2");
    pthread_create(&t3, NULL, route, (void *)"thread 3");
    pthread_create(&t4, NULL, route, (void *)"thread 4");
    pthread_join(t1, NULL);
    pthread_join(t2, NULL);
    pthread_join(t3, NULL);
    pthread_join(t4, NULL);
    return 0;
}

编译运行（这是 C++，用 g++）：

g++ sell_guard.cpp -o sell_guard -pthread
./sell_guard

对比手工版，你发现了吗——代码里一个 unlock 都不用写了。LockGuard lockguard(mutex); 它在构造时上锁，等到这一轮 while 的代码块执行完（无论正常结束还是 break），析构函数就会执行 Unlock()。上一小节提醒你"有多少出口就要解锁多少次"的死角，在 RAII 下自动免疫。

顺带一提：C++ 标准库（C++11 起）也自带了一模一样的两件套——std::mutex 和 std::lock_guard<std::mutex>，用法几乎与上面相同：std::lock_guard<std::mutex> guard(mtx);。我们在这里手写封装的目的，一是能看透它底层到底在调哪些 pthread 接口，二是后续封装线程池、条件变量时能直接复用自己这套。它的细节放到 C++ 课程再展开。

到这里，线程的"互斥"问题我们已经解决得很熟练了。但互斥只解决了一半问题。后半句我们说来就来——线程之间只是"你抢我也抢"还不够，很多时候我们想让线程按预期的顺序配合干活，这就引出了本篇文章的另一根主线：同步。

从互斥到同步：为什么光有锁还不够

先看一个场景，这也是条件变量出生的动机。

假设一个线程要从一个队列里取数据。它加锁、检查队列——发现队列空的。既然空，它就只能"等"。可它怎么等？如果它就这么霸着锁死循环检查（这叫忙等，spin wait），那它占着锁，别的线程想往队列里放数据（放数据也需要拿锁）就永远进不来，最终两个线程互相卡死——一个占着锁等数据，一个想放数据却拿不到锁。这就是"互斥锁解决不了的问题"。

更别扭的是：就算它先把锁放开、等一会儿再回来检查，那它也说不准要等多久，纯靠运气。这显然不是优雅的工程方案。

于是我们引出同步的定义：在保证数据安全（互斥）的前提下，让线程能够按照某种特定的顺序去访问临界资源，从而避免"饿死/忙等"等问题，就叫同步。

顺便把两个词钉死：

• 互斥 sibling：保证"同一时刻只有一个线程进临界区"——解决的是"数据不被破坏"。
• 同步：保证"线程按我们希望的前后顺序去使用资源"——解决的是"谁先谁后、谁等谁醒"。

二者相辅相成。老板和员工之间的"你生产完了，我才能消费"就是一种同步关系。

与同步绑定的另一个词是竞态条件——因为时序（谁先谁后）问题导致程序结果异常，就叫竞态。它和同步是一对"敌人"。

思考题 5：互斥能解决同步问题吗？反过来，同步能替代互斥吗？

答案

都不能。互斥只保证"同时只有一个线程访问共享数据"，不保证"这个线程应该在那个线程之前之后运行"——所以它解决不了同步。反过来，同步强调"顺序配合"，但如果配合过程中对共享容器本身的修改不加锁，两个生产者同时 push 仍可能破坏队列内部结构，所以同步也不能替代互斥。正确姿势是：用互斥保护数据（谁都能改但不能同时改），用同步协调顺序（谁先谁后有约束）。这正是后面生产者-消费者模型里"互斥 + 同步"并用、缺一不可的原因。

条件变量：为"等待"而生

条件变量（condition variable）是 POSIX 提供的一个专门用于线程同步的内核级机制。它的用途一句话就能讲清：当某个线程发现"它需要的条件还不满足"时，它把自己挂起（休眠、释放锁），等别的线程把条件改好了，再来唤醒它。

它解决的问题正是上一小节那个死结：线程占着锁等数据 → 其他线程进不来放数据 → 永远等不到。条件变量通过"等的时候把锁一块交出去"瓦解了死结，我们马上会看到关键接口 pthread_cond_wait 就是干这件事的。

课件里那个直觉例子很形象：一个线程想要访问一个队列，结果发现队列是空的，此时它"什么也做不了"，所以最佳策略不是死等，而是睡一会 + 留个口子让别人把数据放进来，等有人放了、喊醒它，它再来拿。

条件变量的接口

条件变量的类型是 pthread_cond_t，配套函数如下：

逐个讲清楚：

初始化：

int pthread_cond_init(pthread_cond_t *restrict cond,
                      const pthread_condattr_t *restrict attr);

参数 cond 是要初始化的条件变量，attr 传 NULL 用默认属性。也可以像互斥量一样用静态方式：pthread_cond_t cond = PTHREAD_COND_INITIALIZER;。

销毁：

int pthread_cond_destroy(pthread_cond_t *cond);

注意事项和互斥量几乎一样：静态初始化的不用销毁；销毁时要确保没有线程还睡在它上面。

等待条件满足：

int pthread_cond_wait(pthread_cond_t *restrict cond,
                      pthread_mutex_t *restrict mutex);

注意它的签名非常特别——等待条件变量的同时，还要传入一把互斥量。这是本篇文章最重要的一个接口，我们现在先说它的"动作"，后面单独用一大节解释"为什么非要这把锁"。

pthread_cond_wait(cond, mutex) 一进去会做三件事，而且是原子的（一气呵成、中间不会被切换）：

1. 原子地释放你传进来的那把 mutex；
2. 让当前线程挂起在条件变量 cond 上，不再占用 CPU；
3. 直到被唤醒（别人 signal/broadcast）后，它重新去抢回那把 mutex，抢到了才从 pthread_cond_wait 里返回。

唤醒等待者：

int pthread_cond_signal(pthread_cond_t *cond);    // 唤醒一个等待者
int pthread_cond_broadcast(pthread_cond_t *cond); // 唤醒所有等待者

signal 唤醒"一个"（具体是哪个由调度器决定），broadcast 唤醒"全部"。选择哪个，取决于业务：如果资源是"多份、可以多个消费者同时拿"，用 broadcast 合适；如果资源是"单独一份、同时只该有一个处理"，用 signal 即可。

第一个条件变量案例

我们先用最简单的例子感受一下：两个工作线程，各自的"活"要等主线程喊一声才开始。每喊一声 broadcast，两个线程各"活动"一次。

// cond_demo.c   ——  条件变量最初的体验：主线程广播唤醒工作线程
#include <string.h>
#include <unistd.h>
#include <pthread.h>
#include <iostream>
#include <string>
 
pthread_cond_t cond = PTHREAD_COND_INITIALIZER;   // 静态初始化条件变量
pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;// 静态初始化互斥量
 
void *active(void *arg)                            // 工作线程
{
    std::string name = static_cast<const char *>(arg); // 线程名
    while (true)
    {
        pthread_mutex_lock(&mutex);                // 先拿锁
        pthread_cond_wait(&cond, &mutex);          // 等主线程广播；
                                                   // wait 会先释放锁再休眠，
                                                   // 被唤醒后重新抢回锁才返回
        std::cout << name << " 活动..." << std::endl; // 被唤醒后干一次活
        pthread_mutex_unlock(&mutex);              // 干活完解锁
    }
}
 
int main(void)
{
    pthread_t t1, t2;
    pthread_create(&t1, NULL, active, (void *)"thread-1"); // 建两个工作线程
    pthread_create(&t2, NULL, active, (void *)"thread-2");
    sleep(3);                                      // 确保两个线程已经进入等待
 
    while (true)
    {
        // 试试把这一行和下面那行对调，对比效果：
        // pthread_cond_signal(&cond);      // 只唤醒一个线程（注释里演示）
        pthread_cond_broadcast(&cond);     // 唤醒所有在 cond 上等待的线程
        sleep(1);                          // 每秒广播一次
    }
 
    // 上方是死循环，正常永远走不到下面；这里写 join 仅为演示标准写法
    pthread_join(t1, NULL);
    pthread_join(t2, NULL);
    return 0;
}

编译运行：

g++ cond_demo.c -o cond_demo -pthread
./cond_demo

可能的输出（broadcast 每次把两个线程都唤醒）：

thread-1 活动...
thread-2 活动...
thread-1 活动...
thread-2 活动...
thread-1 活动...

如果你把注释里那行 pthread_cond_signal(&cond) 打开、把 broadcast 那行注释掉，就会发现通常一次只唤醒一个线程，"活动"的次数大约只有一半——这正是 signal（唤醒一个）和 broadcast（唤醒全部）的直观区别。

思考题 6：在这个 demo 里，为什么不直接去掉 mutex，只留 cond？wait 前后没有别的条件判断，去掉锁似乎"也能跑"？

答案

这个 demo 是"教学上能跑、工程上经不起推敲"的最小示例。它的 wait 前后没有共享数据需要保护，lock/unlock 在实现细节上看似多余，但这是为了演示"watch-检测-等"的死套路：一是 pthread_cond_wait 的标准接口本来就要求你传入一把合法的互斥量（它内部要原子地释放/重抢这把锁），不给锁接口都没法调用；二是真实工程里"是否要等"的判断（比如 queue 是不是空）一定是守在共享数据上的，你永远需要锁。真正正确的条件变量用法把"判断条件 + 等待"绑定成完整循环，见后面的"使用规范"一节。别被这个 demo 带偏，以为锁可以省。

为什么 pthread_cond_wait 必须要一把互斥量

这是条件变量里最深、也是最重要的一道坎。请把下面这个"错误设计"看仔细，然后我们拆解它为什么错、以及标准接口如何避免它。

一个"想当然"的写法是这样的：先加锁，发现条件不满足，那就"先解锁、再 wait"呗，反正 wait 不就睡了吗。写成代码：

// 错误的设计：先把锁解开，再去 wait
pthread_mutex_lock(&mutex);              // 加锁
while (condition_is_false)               // 条件不满足
{
    pthread_mutex_unlock(&mutex);        // 先解锁
    //   ←--- 危险就藏在这里！
    pthread_cond_wait(&cond, &mutex);    // 解锁之后、等待之前……
    pthread_mutex_lock(&mutex);          // 醒来重新加锁
}
pthread_mutex_unlock(&mutex);

它的问题精确定位在 pthread_mutex_unlock(&mutex) 和 pthread_cond_wait 之间的那个夹缝里（上面对注释处）。这两个动作是分开的、不是原子的。设想时序：

1. 线程 A 解锁（把锁放出去）；
2. 线程 A 还没来得及调用 pthread_cond_wait，就此刻被切换到别处；
3. 线程 B 抢到了锁，把条件改成了"真"，然后 pthread_cond_signal(&cond) 发出唤醒信号；可 A 现在还没睡到 cond 上，这个信号 A 错过；
4. 线程 B 解锁；信号已经发过了；
5. 线程 A 终于回来，调用 pthread_cond_wait 睡到 cond 上……但信号在它睡觉之前就已经错过了，A 永远等不到下一个信号，于是永久阻塞，再也醒不来。

这就是课件强调的：由于解锁和等待不是原子操作，"丢了唤醒信号"导致线程永久卡死。所以"解锁"和"等待"必须是一个原子操作——先释放锁、立刻挂起，中间绝不允许别人插一脚。

pthread_cond_wait(cond, mutex) 正是把"释放 mutex + 挂起到 cond"封装成了一个原子动作，从根上堵死了"夹缝"：要么释放和挂起一起完成，要么都不做，绝不存在"锁已经放了、但人还没睡"的中间态，唤醒信号自然也不可能被错过。

思考题 7：为什么说"释放锁后、立即睡眠"这段必须原子，一旦拆开就会丢信号导致死等？请用"生产者放数据 + 信号"和"消费者解锁后再睡"两个线程的时序，画出一个死锁具体过程。

答案

关键窗口在"消费者已解锁、但尚未 wait"之间。设消费者 C、生产者 P。假若 C 先锁、检查队列空、unlock。正要 wait 的一瞬间，P 抢到锁，往队列 push 一个数据，然后 signal（此时 C 还没睡上去，signal 落空），P 解锁。C 这才执行 wait 睡上去——但唤醒信号早已发出并被错过。此后队列非空、却不该 sync 的 C 也永远不会收到下一个 signal（因为 P 只会在"新放数据"时 signal，而数据已经放进去了、P 不会重复放），C 永久阻塞在 wait 上。破坏这个死锁的唯一办法，就是把 unlock 与挂起焊死成一步，让"放出锁即睡入队"之间不存在任何他人可乘之隙——这正是 pthread_cond_wait 原子完成"释放锁+挂起"的原因。

条件变量的使用规范：为什么判条件是 while 而不是 if

前文分别在两个地方强调过"等待"要对条件做判断，课件也把标准用法总结成了固定套路，我们把它抄下来并逐条解释。

等待条件的线程（消费者这一侧）标准写法：

pthread_mutex_lock(&mutex);                 // ① 拿锁
while (条件为假)                             // ② 用 while，而不是 if！
    pthread_cond_wait(&cond, &mutex);       // ③ 条件不满足就等（自动释放锁、自动重抢锁）
修改条件                                    // ④ 醒来后（条件应满足），更新/消费共享数据
pthread_mutex_unlock(&mutex);               // ⑤ 释放锁

发送条件信号的线程（生产者这一侧）标准写法：

pthread_mutex_lock(&mutex);                 // ① 拿锁
设置条件为真                                // ② 先修改共享数据（把条件改好）
pthread_cond_signal(&cond);                 // ③ 再发信号，唤醒等待者
pthread_mutex_unlock(&mutex);               // ④ 释放锁

两个最多人踩的坑，全部集中在"等待侧"：

坑一：为什么用 while 而不是 if？

如果你写 if (条件不满足) pthread_cond_wait(...)，那么被唤醒后代码会直接继续往下走。可"被唤醒"并不等于"我的条件现在一定满足了"。有两个现实原因：

1. 伪唤醒（spurious wakeup）。POSIX 规定，pthread_cond_wait 即便没有任何线程 signal/broadcast 它，也可能被"虚假地"唤醒。任何一次醒来都必须重新检查条件，用 while 才能兜住。
2. 竞争唤醒。多个等待者共享同一个条件变量，用 broadcast 一次性唤醒全部，但资源只有一份。被唤醒的 A 抢到手把资源消耗掉，B 醒来时资源已经没了——如果 B 用的是 if，它会直接往下访问"其实已经为空/为满"的资源而出错；用 while，B 会重新判断、发现自己还得继续等。

所以正确的挂法是永远用 while (条件) wait，把"醒来就重新检查"写成天然循环。这是教科书级的最标准姿势，没有例外。

坑二：为什么 signal 之前要先设置条件？

因为 pthread_cond_signal 本身只是一个"提醒"，并不搬运数据。真正的"条件变真"是改共享数据（push 元素、task 计数加一）完成的。如果某个线程先 signal、后改数据，另一个被唤醒的线程一看——条件还是没满足——又得睡回去，甚至可能因为丢信号而长眠。所以必须先改条件（改共享数据），再发信号，顺序不能反。这也是上面"发送信号侧"第②③步的顺序依据。

思考题 8：为什么"wait 醒来"之后不能假设条件一定为真？请至少给出"伪唤醒"和"多个消费者竞争一个资源"两个反例。

答案

反例一（伪唤醒）：POSIX 允许条件变量在没有任何 signal 时被系统"意外"唤醒，这是标准刻意的宽松设计——代价是由程序员用再检查来买单。若用 if 简单带过，唤醒后直接消费，而真实数据根本没来，程序就错了。反例二（竞争唤醒）：多个消费者共用一个 cond，生产者用 broadcast 一次唤醒全部，可只生产了一个数据。被唤醒的几个消费者里，先抢到锁的那个把数据取走，其他几个醒来后发现队列已空。它们若用 if 就会越过 empty 判断去取空队列；用 while 则重新检查到 empty，乖乖继续 wait。综上，唯有 while 循环能把"醒来重新验证条件"这一规则网兜住。

生产者消费者模型：为什么需要它

概念铺垫完毕，现在上真正的重头戏——生产者消费者模型。它是多线程工程里最经典、也最实用的一类问题，线程池（我们后面会提）本质上就是它的一个变体。

一句话说清模式

在没有缓冲的时代，生产者和消费者是"你死我活"的强耦合：生产者必须等消费者消化完当前这批数据，才能生产下一批；消费者也必须紧贴着生产者，生产多少就急着处理多少。两边节奏稍有出入，就互相拖后腿。

生产者消费者模型用一个中间"容器"（通常是阻塞队列）把两边隔开：

• 生产者只负责往队列里放数据，放完就走，不等消费者处理；
• 消费者只负责从队列里取数据，取到就处理，不管数据是哪个生产者给的；
• 阻塞队列充当缓冲，平衡两者速度。

双方都不直接和对方通信，全靠队列传话。这就是完整定义，课件原话值得背：

生产者和消费者彼此之间不直接通讯，而通过阻塞队列来进行通讯。生产者生产完数据后不用等待消费者处理，直接扔给阻塞队列；消费者不找生产者要数据，而是直接从阻塞队列里取。阻塞队列相当于一个缓冲区，平衡了生产者和消费者的处理能力，并且把生产者和消费者解耦了。

优点

• 解耦：修改消费者不用动生产者，两套逻辑互相独立。
• 支持并发：生产者生产时消费者可同时消费（在把队列这个共享容器保护好的前提下）。
• 支持忙闲不均：生产者突袭来一大波任务，可以先堆进队列；消费者空闲时慢慢消费，实现"削峰填谷"。谁快谁慢都不会互相卡死。

321 原则

课件用"321"来快速记忆生产消费模型要处理的关系，我帮你展开成完整的：

• 3 种关系：
  1. 生产者 - 消费者：互斥（都碰队列，不能同时）＋ 同步（消费依赖生产，有先后）；
  2. 生产者 - 生产者：互斥（多个生产者不能同时往队列里塞）；
  3. 消费者 - 消费者：互斥（多个消费者不能同时从队列里拿）。
• 2 个角色：生产者和消费者（还要加上那个"交易场所"——共享容器）。
• 1 个交易场所：被大家共用、并被保护起来的队列（容器）。

由此可以直接推演出工程答案：互斥靠一把（或多把）锁，同步靠两个条件变量。为什么要两个条件变量？因为"队列满"和"队列空"是两种要分别等待的信号——生产者等"没满"，消费者等"不空"。

什么时候用互斥、什么时候用条件变量

这一节很多人混。我把决策边界敲清楚：

• 互斥锁（mutex）：管的是"数据"——保证同时只有一个人能改队列。只要涉及对共享容器结构本身的增删改查，就必须要锁。
• 条件变量（cond）：管的是"等待"——保证线程在条件不满足时不忙等、被唤醒时不错乱。当线程发现自己无事可做、需要让别人干完再叫自己时，就得用条件变量，配合 mutex 一起等待。

我们可以给出一个直白的判定：如果并发线程之间只是"各自短暂地碰一下共享数据"，用互斥锁就够；如果存在"某个线程要等另一个线程先把条件改好、而等待期间不能占着锁干等"的情形，就一定得上条件变量，并且要像前文那样"while + wait + lock"地把它配合起来用。后面完整实现里二者同时出现，你对照着看就懂了。

基于阻塞队列的完整实现

我把源码砍成两个文件：一个阻塞队列（头文件），一个主程序。为清楚起见，先以单生产者、单消费者写，注释里我会告诉你"多生产多消费"其实一行都不用改。

先从阻塞队列说起。它和普通队列唯一的不同点，就体现在"阻塞"二字上：

• 队列空时，消费者取元素的操作会被阻塞，直到有元素被放进来；
• 队列满时，生产者放元素的操作会被阻塞，直到有元素被取走。

下面这份 BlockQueue.hpp，我逐行注释，并且把课程里那几个最容易受伤的地方（为什么 wait 要用 while、为什么 wait 前要记录等待者数量、为什么在临界区里 wait 不会死锁、signal 前为什么要先判断有没有等待者）全部标出来：

// BlockQueue.hpp  ——  用 std::queue 模拟"线程安全 + 可阻塞"的队列
#ifndef __BLOCK_QUEUE_HPP__
#define __BLOCK_QUEUE_HPP__
 
#include <iostream>
#include <queue>        // std::queue 标准队列容器
#include <pthread.h>    // pthread 线程原语
 
template <typename T>          // 模板：队列里不仅能放 int，也能放任意类型的"任务对象"
class BlockQueue
{
private:
    bool IsFull()                        // 判断队列是否已满
    {
        return _block_queue.size() == _cap;   // 元素个数达到容量上限即满
    }
    bool IsEmpty()                       // 判断队列是否为空
    {
        return _block_queue.empty();          // 容器的 empty 即空
    }
 
public:
    // 构造函数：cap 是容量上限，并初始化锁和两个条件变量
    BlockQueue(int cap) : _cap(cap)
    {
        _productor_wait_num = 0;             // 当前睡着等待"有空位"的生产者个数
        _consumer_wait_num = 0;              // 当前睡着等待"有数据"的消费者个数
        pthread_mutex_init(&_mutex, nullptr);       // 初始化保护队列的锁
        pthread_cond_init(&_product_cond, nullptr); // 给生产者用：等待"有空位"
        pthread_cond_init(&_consum_cond, nullptr);  // 给消费者用：等待"有数据"
    }
 
    // 生产者调用：往队列里放一个元素
    void Enqueue(T &in)
    {
        pthread_mutex_lock(&_mutex);         // 进临界区，拿锁保护队列
        while (IsFull())                     // 队列满了吗？（用 while 不是 if！）
        {
            // 满，生产不下去 → 睡在这里等"空位"。注意：此时我们还拿着锁呢！！
            //
            //   ★ 关键：pthread_cond_wait 这一步做了三件事（原子完成）：
            //     1) 让当前线程挂起在 _product_cond 上（睡）
            //     2) 自动释放我们手上这把 _mutex 锁（把锁交出去）
            //     3) 被唤醒后，先重新抢回 _mutex 锁，抢到了才从这里返回
            //
            //   所以在 wait 期间锁是"被让出去的"，其他生产者/消费者能拿锁干活，
            //   不会因为"睡在临界区里"而互相死锁。这正是 wait 设计成要传锁的原因。
            _productor_wait_num++;           // 记录一下"我这个生产者醒着要来等"
            pthread_cond_wait(&_product_cond, &_mutex); // 睡！（会自动解锁）
            _productor_wait_num--;           // 醒来后，我就不再算等待者了
            // 醒来后回到 while 重新判满：可能会有人已经放进来→再也不满，才继续
        }
        // 能走到这里，说明此刻队列必然不满（while 已经保证）→ 真正执行生产
        _block_queue.push(in);               // 把元素放入队列
        if (_consumer_wait_num > 0)          // 如果有消费者正睡着等数据
            pthread_cond_signal(&_consum_cond); // 就唤醒一个（signal 即可）
        pthread_mutex_unlock(&_mutex);       // 生产完，释放锁
    }
 
    // 消费者调用：从队列里取出一个元素到 out
    void Pop(T *out)
    {
        pthread_mutex_lock(&_mutex);         // 进临界区，拿锁
        while (IsEmpty())                    // 队列空了吗？（用 while 不是 if！）
        {
            // 空，没得消费 → 睡在这里等"有数据"。原理同上：
            // wait 会原子地释放锁再睡，被唤醒后重新抢锁再返回。
            _consumer_wait_num++;            // 记录我这个消费者在等
            pthread_cond_wait(&_consum_cond, &_mutex); // 睡！（会自动解锁）
            _consumer_wait_num--;            // 醒来后不再算等待者
            // 醒来后回到 while 重新判空（治伪唤醒 + 竞争唤醒）
        }
        // 此刻队列必然非空 → 真正执行消费
        *out = _block_queue.front();         // 取队头元素
        _block_queue.pop();                  // 把它从队列里删掉
        if (_productor_wait_num > 0)         // 如果有生产者正睡着等空位
            pthread_cond_signal(&_product_cond); // 唤醒一个
        pthread_mutex_unlock(&_mutex);       // 消费完，释放锁
    }
 
    // 析构：销毁所有同步原语
    ~BlockQueue()
    {
        pthread_mutex_destroy(&_mutex);        // 销毁互斥量
        pthread_cond_destroy(&_product_cond);  // 销毁生产条件变量
        pthread_cond_destroy(&_consum_cond);   // 销毁消费条件变量
    }
 
private:
    std::queue<T> _block_queue;    // 底层真正存的队列容器（是被整体保护的共享资源）
    int _cap;                      // 队列容量上限
    pthread_mutex_t _mutex;        // 保护 _block_queue 的互斥锁
    pthread_cond_t _product_cond;  // 给生产者用的条件变量：等待"队列有空位"
    pthread_cond_t _consum_cond;   // 给消费者用的条件变量：等待"队列有数据"
    int _productor_wait_num;       // 当前正睡着等待空位的生产者个数
    int _consumer_wait_num;        // 当前正睡着等待数据的消费者个数
};
 
#endif // __BLOCK_QUEUE_HPP__

几个你必须反复咀嚼的设计点：

第一，wait 期间虽然"代码还在临界区里、锁还被我攥着"，但它会自动放锁。这是 pthread_cond_wait 的看家本领。它返回时又自动把锁抢回来，所以 wait 前后你都不需要手动 lock/unlock。这正是我们上一大节反复强调的"释放锁 + 睡眠"必须原子，否则会丢信号。

第二，醒来后用 while 重新判条件，而不是直接往下走。这一方面兜住"伪唤醒"（没 signal 也会醒），另一方面兜住"我醒了但资源被别人抢走"的竞争。所以 while(IsFull())、while(IsEmpty()) 这两句，一句都不能省、不能改成 if。

第三，_productor_wait_num / _consumer_wait_num 这两个计数的作用是：在发信号前判断"有没有人真的在等"。如果没有任何生产者等待空位，你 signal(&_product_cond) 就是白喊一声（甚至没必要）；反过来，有消费者在等数据时你就该唤醒。用计数判断"要不要 signal"，能让信号发的有的放矢、减少无谓唤醒。

第四，生产者和消费者分别用两个不同的条件变量，互不错唤：生产者只等 _product_cond（空位），消费者只等 _consum_cond（数据）。这样 broadcast/signal 都精准，不会出现"把生产者错喊起来结果发现仍没空位"的浪费。

接下来写主程序 producer_consumer.cpp。为了让"多生产多消费"也能演示，我直接创建多对线程。模板参数用 int，方便理解——课件提醒过，改成任务对象（比如 std::function<void()>）队列照样能用，因为它是模板。

// producer_consumer.cpp   ——  用阻塞队列搭一个生产者消费者模型
#include <iostream>
#include <unistd.h>           // sleep/usleep
#include <pthread.h>
#include "BlockQueue.hpp"     // 引入阻塞队列
 
using namespace std;
 
// 消费者入口：不断地从队列取数据并"消费"
void *consumer(void *args)
{
    // 通过 void* 把 BlockQueue<int>* 传进来，再还原成指针使用
    // （用 void* 是为了满足 pthread_create 的回调签名要求）
    BlockQueue<int> *bq = static_cast<BlockQueue<int> *>(args);
    while (true)
    {
        int data = 0;
        bq->Pop(&data);                  // 取数据；若队列空，Pop 内部会阻塞等待
        cout << "消费了一个整数: " << data << endl; // 模拟"处理数据"
        sleep(1);                        // 消费者慢一点，制造"忙闲不均"
    }
    return NULL;
}
 
// 生产者入口：不断地生产整数放队列
void *productor(void *args)
{
    BlockQueue<int> *bq = static_cast<BlockQueue<int> *>(args);
    int cnt = 0;
    while (true)
    {
        bq->Enqueue((cnt++));            // 把递增的整数放进队列；若满，Enqueue 内部阻塞
        cout << "生产了一个整数: " << cnt << endl;
        usleep(200000);                  // 生产者快一点（200ms 一个），体现缓冲作用
    }
    return NULL;
}
 
int main(void)
{
    BlockQueue<int> bq(5);               // 建一个容量为 5 的整数阻塞队列
    pthread_t p[3], c[3];                // 3 个生产者线程、3 个消费者线程的 id
 
    for (int i = 0; i < 3; ++i)          // 建 3 个生产者
        pthread_create(&p[i], NULL, productor, &bq);
    for (int i = 0; i < 3; ++i)          // 建 3 个消费者
        pthread_create(&c[i], NULL, consumer, &bq);
 
    // 阻塞等待所有线程结束（生产/消费都是死循环，正常会一直跑）
    for (int i = 0; i < 3; ++i) pthread_join(p[i], NULL);
    for (int i = 0; i < 3; ++i) pthread_join(c[i], NULL);
    return 0;
}

编译运行：

g++ producer_consumer.cpp -o producer_consumer -pthread
./producer_consumer

你会看到生产/消费交错进行，队列容量 5 被用满就"满生产阻塞、空消费阻塞"，但绝不会出现"队列被同时篡改"或"负数/越界"。这个模型里的锁保证了数据安全，两个条件变量保证了顺序配合——互斥与同步在我们的代码里同时工作，这正是本文章两条主线第一次真正汇聚。

注意：Enqueue 的形参我写的是 T &in（引用），所以调用处要传变量 cnt++ 这种可修改的左值。如果你想把接口改成 Enqueue(const T &in) 甚至 Enqueue(T &&in)（移动语义），也完全合法，详见你学的 C++ 内容。

思考题 9：把阻塞队列里两个 while 改成 if，在"多个消费者 + 一个生产者 + broadcast 唤醒"的场景下，会出什么问题？请描述具体的坏结果。

答案

假设队列同时有多个消费者等在空条件 + 睡眠，生产者放一个数据后 signal/broadcast 一次唤醒多个。若用 if，被唤醒的消费者直接执行 front() + pop()，可数据只有一份——A 抢到锁取走了数据，B 醒来后（它已拿回锁、若抢先）队列事实上已空，却仍直接 pop，轻则取到脏数据，重则对空队列操作导致未定义行为（崩溃、越界）。而用 while，B 醒来后重新判 IsEmpty() 为真，愉快地继续睡，安全无事。由于"broadcast 唤醒多个 + 资源只有一份"的组合在工程中很常见，所以 if 的判断在这里是必踩的坑。

POSIX 信号量：另一种同步工具

讲完条件变量，课件紧接着介绍了 POSIX 信号量。很多教材直接用信号量重写生产者消费者，它和条件变量的思路不同，但对"同步"这一需求而言是同一类工具——都用来协调线程顺序。我们在这一节把它讲清楚，并在结尾和条件变量做个对比。

信号量是什么

信号量（semaphore）是一个维护"整数值"的计数器。两个核心操作让人印象深刻，直接用了荷兰语的缩写：

• sem_wait，又称 P 操作：把信号量的值减一。如果减完小于 0（或尚未大于 0），当前线程阻塞，直到有别的线程让它醒来；若减后仍 ≥ 0，立即返回继续。
• sem_post，又称 V 操作：把信号量的值加一。如果有线程正因等待某个值被阻塞，会唤醒其中之一。

语义上：信号量的值通常表示"当前还有多少可用资源"。这样：

• 要"占用一份资源"，就 sem_wait（拿一份，值减一）；
• 要用完归还，就 sem_post（还一份，值加一）；
• 值为 0 时再 wait，表示资源没了，线程得等到别人 post。

和互斥量最大的区别是：信号量的值可以大于 1，天生能表达"多个份的资源"；而互斥量只有 0/1，只能表达"一个进出标志"。所以同样是保护资源，互斥量"独占地进一个"，信号量"限量地进多个"。

接口

#include <semaphore.h>
 
int sem_init(sem_t *sem, int pshared, unsigned int value);
//  pshared：0 表示线程间共享；非 0 表示进程间共享
//  value  ：信号量的初始值（资源初始份数）
//  成功返回 0，失败返回 -1（注意：信号量失败返回 -1 并设置 errno，与 pthread 接口不一致）
 
int sem_destroy(sem_t *sem);        // 销毁信号量
 
int sem_wait(sem_t *sem);           // P 操作：值减 1，若减完 < 0 则阻塞
int sem_post(sem_t *sem);           // V 操作：值加 1，并唤醒一个等待者

用信号量实现环形队列的生产者消费者

阻塞队列用 std::queue，空间动态分配。现在我们用固定大小的环状队列（数组模拟 + 取模模拟"环"）+ 信号量重写一遍。环状数组有个天然难点：满和空时游标位置一样，不好区分。但信号量恰好提供了天然计数器——用两个信号量分别管理"剩余空间数"和"已有数据数"，就无需再用标记去分辨满/空。

先造一个信号量的小封装 Sem.hpp：

// Sem.hpp   ——  对 POSIX 信号量的薄封装
#pragma once
#include <semaphore.h>      // 信号量接口
 
class Sem
{
public:
    Sem(int n)                              // 构造：初始值 n
    {
        sem_init(&_sem, 0, n);              // pshared=0 → 线程间共享
    }
 
    void P()                                // P 操作：拿一份资源（值 -1，不够则阻塞）
    {
        sem_wait(&_sem);
    }
 
    void V()                                // V 操作：归还一份资源（值 +1）
    {
        sem_post(&_sem);
    }
 
    ~Sem()                                  // 析构：销毁
    {
        sem_destroy(&_sem);
    }
 
private:
    sem_t _sem;                             // 底层信号量
};

再写环形队列 RingQueue.hpp。单生产者单消费时甚至不需要锁，因为"一个写一个读"天然不会冲突；但为支持多生产多消费，课件给生产者和消费者各配了一把锁维护互斥：

// RingQueue.hpp   ——  基于固定大小环形数组 + 信号量的生产消费模型
#pragma once
#include <vector>
#include <semaphore.h>
#include <pthread.h>
#include "Sem.hpp"
 
template <typename T>
class RingQueue
{
private:
    void Lock(pthread_mutex_t &mutex)       // 加锁的小封装
    {
        pthread_mutex_lock(&mutex);
    }
    void Unlock(pthread_mutex_t &mutex)     // 解锁的小封装
    {
        pthread_mutex_unlock(&mutex);
    }
 
public:
    // 构造：cap 是容量；_room_sem 初始 = cap（空位数），_data_sem 初始 = 0（数据数）
    RingQueue(int cap)
        : _ring_queue(cap),
          _cap(cap),
          _room_sem(cap),      // 空位信号量：一开始有 cap 个空位
          _data_sem(0),        // 数据信号量：一开始有 0 份数据
          _productor_step(0),  // 生产者下次写入的下标
          _consumer_step(0)    // 消费者下次取出的下标
    {
        pthread_mutex_init(&_productor_mutex, nullptr); // 保护所有生产者的互斥
        pthread_mutex_init(&_consumer_mutex, nullptr);  // 保护所有消费者的互斥
    }
 
    void Enqueue(const T &in)               // 生产者入口
    {
        _room_sem.P();                      // ★ 先申请一个"空位"；没有空位就阻塞在这
        Lock(_productor_mutex);             // 多生产者互斥：同一时刻只一个在生产
        _ring_queue[_productor_step++] = in;// 写入当前生产位置
        _productor_step %= _cap;            // 用取模把下标卷回 0，模拟"环"
        Unlock(_productor_mutex);           // 释放生产锁
        _data_sem.V();                      // ★ 生产了一块数据，数据信号量 +1
    }
 
    void Pop(T *out)                        // 消费者入口
    {
        _data_sem.P();                      // ★ 先申请一份"数据"；没有就阻塞在这
        Lock(_consumer_mutex);              // 多消费者互斥
        *out = _ring_queue[_consumer_step++]; // 从当前消费位置取走一份
        _consumer_step %= _cap;             // 用取模卷回，模拟环
        Unlock(_consumer_mutex);            // 释放消费锁
        _room_sem.V();                      // ★ 消费了一块，归还一个"空位"
    }
 
    ~RingQueue()                            // 析构：销毁两把锁
    {
        pthread_mutex_destroy(&_productor_mutex);
        pthread_mutex_destroy(&_consumer_mutex);
    }
 
private:
    std::vector<T> _ring_queue;          // 环形数组（容量固定 _cap）
    int _cap;                            // 容量上限
    int _productor_step;                 // 生产游标
    int _consumer_step;                  // 消费游标
    Sem _room_sem;                       // 空位信号量：生产者关心"有没有空位"
    Sem _data_sem;                       // 数据信号量：消费者关心"有没有数据"
    pthread_mutex_t _productor_mutex;    // 生产者之间的互斥锁
    pthread_mutex_t _consumer_mutex;     // 消费者之间的互斥锁
};

注意这里锁和信号量的配合很有意思：信号量管"能不能"（顺序/数量），锁管"同时能不能"（互斥）。_room_sem 控制"生产者不能超过 cap 个、空位不够时就等着"，_data_sem 控制"消费者没数据时只能等"——这正是"同步"；两把锁再兜底"就算能生产，同一时刻也只能一个生产者写"——这正是"互斥"。同步 + 互斥，又一次同时在场。

思考题 10：sem_wait（P）和 sem_post（V）能不能像 ticket-- 那样被非原子地打断？信号量的值维护会不会也出竞态？

答案

不会。sem_wait/sem_post 的实现底层同样借助了原子操作（如原子比较交换 CAS）来对计数器做增减，标准库已经保证了信号量的 P/V 操作自身是原子的。也就是说，"检测值是否可减"和"真的去减"绑定成了一个原子动作，不存在中间被别的线程插一脚的窗口。与 ticket--（三条指令、非原子）截然不同，信号量把"检查 + 修改"焊死成一体，因此能安全地用于并发计数。这一点正是它能替代手写计数器当"可用的资源数"来用的底气。

条件变量 vs 信号量：怎么选

把两者放在一张表里，你就能记住何时用谁：

一句话总结：条件变量擅长"复杂条件 + 需要记录是否在等待"，信号量擅长"简单资源计数 + 天然带数值"。生产消费模型两者都能实现，条件变量写法更通用可控，信号量写法更精简但语义较粗糙。工程上两者都常见，是同一道题的两种解法。

死锁：线程们互相"拿瓶盖"

铺垫了这么多同步工具，如果使用不当，就会产生一种非常阴险的故障——死锁。它不像竞态那样跑着跑着出错然后崩溃，而是会让程序"静悄悄地卡死"，CPU 占用为 0，仿佛时间停止，却怎么都结束不了。

死锁的定义是这样的：一组执行流（线程/进程）中，每个都占有一份"不会释放的资源"，同时又都在申请被别的执行流占有的、也不会释放的资源，于是在这个环形链条里，每个人都拿着自己那份、等着别人那份，互相等待、谁也无法前进，进入一种"永久等待"的状态。

用课件那个桩子：假设线程 A、线程 B 要同时持有锁 1 和锁 2 才能继续干活，但——

• 线程 A 先拿到了锁 1，然后去申请锁 2；
• 线程 B 先拿到了锁 2，然后去申请锁 1；
• A 等 B 释放锁 2，B 等 A 释放锁 1。

两人手里都攥着一个锁，谁也不放，谁也等不到对方，于是双双挂死。这个画面想象成两个小朋友：每个人都用两只手使劲拽一只瓶盖，可两个人拽的是同一个瓶子，拧到一半谁也奈何不了谁。"申请一把锁是原子的；申请两把锁就不一定了"——这句话精准地指认了危险所在。

死锁的四个必要条件

死锁的成因被经典地归纳成四个必要条件，缺一不可。理解这四个条件，最大的价值在于：只要破坏其中任何一个，死锁就不可能发生，这直接给了我们避免死锁的抓手。

1. 互斥条件：一个资源每次只能被一个执行流使用。任何时刻，这个资源属于且只属于一条执行流，别人想用只能等。（互斥是这种资源的天性，无法轻易破除。）
2. 请求与保持条件：一个执行流在"请求新的资源"的等待过程中，对已经获得的资源绝不放弃，始终攥着不放。（等待的时候不撒手。）
3. 不可剥夺条件：一个执行流已经获得的资源，在它用完之前，不能被其他线程强行夺走。资源只能由持有者自己主动释放。
4. 循环等待条件：若干执行流之间，形成一种头尾相接的循环等待链路：A 等着 B 手里的资源，B 等着 C 手里的，C 又等着 A 手里的……正好绕成一个环。

思考题 11：有人说"只要打破死锁四个必要条件之一，就绝对不会死锁"。请问最容易被人为改写、也最常被程序员利用来破局的是哪一个？为什么？

答案

最常被利用的是第 4 条"循环等待条件"。理由：互斥条件（资源已占用性质）根植于资源本身，不可剥夺条件需要硬件/调度器提供强占能力（代价大、很少用），请求与保持条件可以通过"先把想要的所有资源一次性全部申请到再干活"来破除但工程上往往不合适；而循环等待——只要规定"所有线程按同一固定的全局顺序加锁"，就能彻底消除环。因为互斥、不可剥夺是资源与调度器决定的 "硬条件"，程序员能高性价比干预的，基本就是"调整加锁顺序来打断环"这一条。所以工程上"全局统一加锁顺序"成了避免死锁的头号手段。

怎么避免死锁

诚如上面的思考题所揭示的：避免死锁的核心思路，就是消灭四个必要条件之一。而现实中最好操作的，通常是打破循环等待。结合课件给出了下面几条实战手段：

手段一：加锁顺序一致（打破循环等待）。规定所有线程都必须遵循同一个加锁顺序，比如"永远先锁 1 再锁 2，不允许反过来"。这样就不可能出现"A 持 1 要 2、B 持 2 要 1"的环，因为谁都不会先去锁 2—环被切断了。这是系统里最简单有效的一招。

手段二：资源一次性分配 / 加锁超时。要么一次性把所有需要的资源都拿到手，要么用带超时的加锁（如 C++ 里 std::timed_mutex 的 try_lock_for，或 pthread 的 pthread_mutex_timedlock），拿不到就撤，不让"请求与保持"无限持续。带超时还能顺带"自愈"：拿不到就放弃已持有的、过会儿再重试，破坏"请求与保持 + 不可剥夺"这个组合。

手段三：用标准库的 std::lock 同时锁多把互斥量。C++11 提供了 std::lock(mtx1, mtx2, ...)，它内部通过"尝试-成功则上、失败则回滚重试"的方式，保证一次性无环地获取多把锁，从机制上绕过死锁。课件里提供了这样一个示例骨架，我整理成可运行的完整代码：

// deadlock_avoid.cpp   ——  对比"分开加锁"与"一把同时加"对结果的影响
#include <iostream>
#include <mutex>
#include <thread>
#include <vector>
 
int shared_resource1 = 0;          // 共享资源 1
int shared_resource2 = 0;          // 共享资源 2
std::mutex mtx1, mtx2;             // 保护这两个资源的互斥量
 
// 一个函数，同时"访问"两个共享资源
void access_shared_resources()
{
    // ====== 写法 A：分别加锁（容易死锁，尤其多线程时）======
    // mtx1.lock();
    // mtx2.lock();
    // // 临界区……（为演示，直接循环累加）
    // for (int i = 0; i < 10000; ++i) { ++shared_resource1; ++shared_resource2; }
    // mtx1.unlock();
    // mtx2.unlock();
 
    // ====== 写法 B（推荐）：一次性、无环地同时锁定两把锁 ======
    std::unique_lock<std::mutex> lock1(mtx1, std::defer_lock); // 先创建但不立即加锁
    std::unique_lock<std::mutex> lock2(mtx2, std::defer_lock); // 同样延迟加锁
    std::lock(lock1, lock2);  // 一次性同时锁住两个互斥量，自动避免循环等待
    // 之后安全地访问两个共享资源
    for (int i = 0; i < 10000; ++i)
    {
        ++shared_resource1;
        ++shared_resource2;
    }
    // 离开作用域时 lock1/lock2 析构，自动解锁
}
 
int main()
{
    std::vector<std::thread> threads;
    for (int i = 0; i < 10; ++i)           // 10 个线程并发访问
        threads.emplace_back(access_shared_resources);
    for (auto &t : threads)                // 等待全部完成
        t.join();
    // 正确结果应是 100000 / 100000（10 个线程各加 10000）
    std::cout << "Shared Resource 1: " << shared_resource1 << std::endl;
    std::cout << "Shared Resource 2: " << shared_resource2 << std::endl;
    return 0;
}

编译运行：

g++ deadlock_avoid.cpp -o deadlock_avoid -pthread
./deadlock_avoid

用写法 B（std::lock）一次锁定，输出稳定为两个 100000（每个线程各加 10000，10 个线程正好 100000×2）。若换成写法 A 那种"先 lock mtx1、再 lock mtx2"而另一个线程可能反过来先 mtx2 再 mtx1，就可能在多线程不断切换中出现死锁，数值跑不齐。

关于死锁的"检测"与"银行家算法"：操作系统课程还有专门的死锁检测和银行家算法等经典课题，这是资源分配管理层面的主题，超出了本文范围。你只需要知道它们的存在，若日后学习操作系统原理再深入——它们的本质，都是围绕那四个必要条件想方设法做事前规避或事后恢复。

线程安全与可重入

最后两个高频概念：线程安全和可重入。它们经常被混为一谈，其实是两个不同层面的问题。我把各自的定义、判定、常见情况、以及二者关系一次性讲透。

什么是线程安全

线程安全：多个线程并发地去访问同一个代码/同一份共享资源时，能正确执行、不相互干扰、不破坏彼此的运行结果。这说的是一种"从多线程外部看数据依然一致"的性质。

直观判据：如果一段并发 code，无论线程怎么交错执行，结果都和我们想象的一致，就说明它是线程安全的。

反例就是我们的售票系统——多个线程卖票，结果卖到负数，所以那个不加锁的 route 就不是线程安全的。

什么是可重入

可重入：同一个函数被不同的执行流调用，而前一个执行流还没执行完、后一个执行流就又进来了（函数"重入"了），在这种情况下函数运行结果依然正确，那么它就是可重入函数；否则是不可重入函数。

注意 "重入" 有两种来源（课件专门点了名）：

• 多线程重入：两个线程同时执行同一个函数（这个我们一直在谈）；
• 信号导致的重入：一个执行流正在执行函数 F 时，来了一个信号，信号处理函数里又调用了 F，于是同一个执行流没执行完就"再次进入"了同一个函数。

线程不安全 vs 不可重入：常见情形对照

把课件里的结论整理成表，比死记概念有用得多：

线程安全与可重入的联系与区别

这一点是很多题的考点，也是理解的重灾区。我把课件结论整理成三条"一句话"和一张关系表：

联系（课件原话）：

1. 函数若可重入，那它就是线程安全的（记住这一句就够了，它省去了无数纠结）；
2. 函数若不可重入，就不能由多个线程使用，否则可能引发线程安全问题；
3. 函数里若有全局变量，那它通常既非线程安全也非可重入。

区别：

• 可重入函数是"线程安全函数"的一种——可重入 ⊂ 线程安全；
• 线程安全不一定可重入，但可重入一定线程安全；
• 关键差异：如果把对临界资源的访问加上了锁，这个函数是线程安全的；可这个加锁的可重入函数如果锁还没释放就再次进入，就会死锁，因此这个函数不可重入（比如最典型的——把线程安全函数改成可重入函数时，那个不可重入部分恰恰是锁：锁没释放时第二次进入同一把锁 = 死锁）。

可重入 ⊆ 线程安全（即：可重入函数必然是线程安全的，
       但线程安全函数不一定是可重入的）

思考题 12：给一个用了互斥锁保证线程安全的函数，它为什么可能"可重入"失败？举个典型例子。

答案

因为锁本身是"不可重入"的：同一个线程对同一把普通互斥锁连续调用两次 lock（第一次还没 unlock，第二次又 lock），会直接死锁。典型的例子是一个内部用锁保护全局计数器的函数 inc()，在临界区里调用它自己（递归）——外层已经持锁，内层再 lock 同一把锁，就是"主动重入同一把未释放的锁"，立刻自锁死。这也解释了为什么 C++ 还专门提供"递归锁"（std::recursive_mutex/pthread_mutex_recursive）：它允许同一个线程重复加锁而不死锁，以换回这种场景下的"可重入"。注意这属于属性定制，普通互斥默认不给。

STL 与智能指针的线程安全吗

学完了线程安全，你自然会关心一个更实际的问题：我用得最多的 C++ 标准库容器、智能指针，它们线程安全吗？课件给了一个非常务实的结论，我复述并补充如下。

问题一：STL 容器（vector、list、map……）线程安全吗？

回答：默认不是。

原因一句话：STL 的设计目标是"把性能榨到极致"，而一旦为每个容器内置锁，性能会受到巨大影响；且不同容器对锁的分摊策略不同（比如 hash 表可以"锁表"或"细化到锁每个桶"，性能又不一样），标准库选了"不内置锁、由调用者负责同步"这条最干净的路。

所以，多线程下要共享一个 std::vector，你需要自己给它配一把锁（或像我前面封装的 Mutex + LockGuard 那样包起来），自行确保读写时互斥。

问题二：智能指针线程安全吗？

要分两类看：

• unique_ptr：它是"独占所有权"，一块对象同时只有一个智能指针持有，其生命周期只在当前作用域内，基本不涉及线程安全问题（你不共享它，自然谈不上并发冲突）。
• shared_ptr：多个 shared_ptr 共享同一个引用计数变量。引用计数的增减如果被并发执行，窗口一开就可能出现计数错乱（提前销毁或永不销毁）。标准库在实现时已经考虑到了：引用计数的增减是基于原子操作（如 CAS）实现的，所以多个线程安全地增减引用计数、安全地拷贝/销毁 shared_ptr 是可靠的。

但有一个更细的坑要提：shared_ptr 的"对象本身"它不管。引用计数线程安全 ≠ 它所指向的那个对象线程安全。两个线程同时读同一个 shared_ptr 指向的对象、并各自去改它，那依然是要我们自己加锁的。别把"引用计数安全"错当成了"对象安全"。

思考题 13：shared_ptr 引用计数是原子的，那是不是意味着把同一个 shared_ptr 的裸指针丢给多个线程随便用也安全？

答案

不是。原子性只保证了"引用计数的增减"这一件事在线程间安全（从而不会提前释放、不会泄漏），它不保证"所指对象内部的数据"安全。多个线程通过同一个裸指针并发地去修改对象成员，如果不加锁，属于对同一份内存的普通数据竞争，照旧会出竞态问题。所以即便 shared_ptr 引用计数是原子的，共享对象本身仍然要我们自觉加锁保护。一句话：原子引用计数解决"内存生命周期"问题，不解决"对象内容并发访问"问题。

其他常见锁：悲观乐观、CAS、自旋与读写

最后把散落在各种框架里的"锁"名词归个类，让你在别的资料或面试里看到它们时不慌。这些大多是"同一把锁在不同场景的变体",核心思想都和你已经掌握的一样。

• 悲观锁：每次取数据之前，总是"担心数据会被别人改掉"，所以先加锁再读取/修改，整个访问期间锁一直持有，其他线程想访问就被阻塞挂起。数据库里的行锁、表锁就是典型。它的代价是：明明大多数时候没有冲突，也先锁了，性能有损耗（锁开销 + 串行化）。
• 乐观锁：每次取数据时"乐观地"认为没人会改它，因此不加锁直接读。但在真正写入（更新）之前，会去核对数据有没有被别人改过（这就是"冲突检测"）。检测手段有两种主流：
  1. 版本号机制：数据带一个版本号，更新时核对版本号没变才更新，并把版本号 +1；若发现版本号已变，说明别人改过了，本次更新失败重试。
  2. CAS 操作（见下）。
• CAS（Compare And Swap）：一条原子指令。做更新时，先比较"当前内存值"和"之前取到的值"是否相等：相等，说明没人动过，就用新值替换；不等，说明被改过，本次失败。失败后通常进入一个重试循环（自旋），直到成功为止。它实现了一种"无锁"的乐观更新，性能远高于悲观锁，是很多无锁数据结构的根基。
• 自旋锁：等待锁时不挂起线程，而是原地空转（不断循环尝试拿锁）。好处是避免了线程切换的开销；坏处是临界区长时白白烧 CPU。适合临界区极短的场景。
• 读写锁：区分"读者"和"写者"。允许多个读者同时进入（读读不互斥）、但读者和写者、写者和写者之间互斥（读写互斥、写写互斥）。适合"读多写少"的场景，能大幅提升并发度。

你可以把这几种锁，看作关于"何时加锁、为何等待、怎么等待"的三组对位：

这篇文章我们从"多线程卖票卖出负数"这个看似简单的 bug 出发，一路拆穿了它背后"读-改-写三态非原子"的真相；然后用互斥量给它上了锁，透过 CAS 看清了这把锁为何自己也能保持原子；又顺着"互斥只能防数据破坏、管不了干活顺序"这条线索，引出了条件变量，把 pthread_cond_wait 为什么必须配互斥量、为什么等待要用 while 这些深坑讲了个明白。接着我们用阻塞队列完整实现了生产者消费者模型，又介绍了信号量这一同类工具，最后系统地过了一遍死锁的四个必要条件、线程安全与可重入的异同，以及乐观/悲观、CAS、自旋、读写等拓展锁概念。

一句话收官整篇文章：互斥负责"让并发不出错"，同步负责"让并发有秩序"；学习的终点，是你既能用锁把共享数据守得密不透风，又能用条件变量让线程配合得行云流水。 把这篇文章里每个例子亲手编译跑一遍、把每个"思考题"想一遍，线程同步这门手艺就算真正上手了。

下一篇文章，我们会把这里封装的锁、条件变量和线程，组合成"线程池"这个生产环境里高频使用的利器，你会在那里看到所有这些概念是如何拧成一股绳、扛起真实并发任务的。