生产消费模型实战：用C语言讲透互斥锁与条件变量

两个线程同时执行count连续跑一万次最后 count 一定等于两万吗相信写过并发代码的人都会下意识摇头但你要是追问一句“为什么”很多人就开始含糊了。线程同步、线程互斥、生产消费模型这三个词单独看都能说上两句可一旦要把它们组合起来写出一段能稳定运行的程序就完全是另一回事。这篇内容就是要从最朴素的多线程问题出发把它引到生产消费模型上再用完整代码把互斥锁和条件变量讲透。适合正在学操作系统、学并发编程或者写过多线程但被各种随机 Bug 折磨过的同学有基础的人也能从原理层面再拉一拉认知。1. 生产消费模型到底在解决什么问题1.1 从一次事故聊起变量自增其实有三步很多初学者第一次翻车就是翻在count上。你以为是“一步操作”但在 CPU 眼里它是“读内存、加一、写回内存”三步。两个线程 A 和 B 同时执行这三步完全可能出现这种交错A 读取到 count 0B 读取到 count 0A 加一后写回count 1B 加一后写回count 1最终 count 只加了 1而不是 2。这还不算最狠的更狠的场景是多个线程同时往一个链表里插入节点插入到一半被切走另一个线程遍历链表直接崩溃。我当年第一次跑出这种结果时第一反应是编译器出 Bug 了反复验证大半天才发现问题出在“没有做线程同步”。多个线程在没有任何控制的情况下同时访问共享资源产生的结果依赖于线程调度的先后顺序而不是依赖代码本身的逻辑。这个现象有一个专门的名字竞态条件。发生竞态的那段访问共享资源的代码就是临界区。想解决竞态核心思想只有两个字互斥。共享资源就像只有一个工位的厨房你不能让两个厨师同时占着这个工位切菜必须规定谁先拿到工位牌谁进去做完再交出来后面的人才能进。1.2 生产者和消费者本质是两类线程的协作生产消费模型其实不是某个高深的理论它就是把现实里的协作关系抽象成两类线程。一类线程负责产出数据叫生产者另一类线程负责处理数据叫消费者中间夹着一个缓冲区用来临时存放数据。为什么要中间夹缓冲区最直白的理由就两个字解耦。生产者不需要知道消费者的处理速度消费者也不需要等生产者现做两边各干各的节奏不一致也没关系。就像面包店里的后厨和前台之间永远有一层货架后厨烤完就往货架上放前台有客人就过来取谁也不用盯着谁干活。但货架是有容量上限的。一旦放大规模后厨有两个人同时烤面包前台也有两个人同时卖货问题就来了两个后厨不能同时把面包往同一个货架格子里放不然位置会覆盖这就是互斥问题货架满了后厨必须暂停生产等前台卖掉一些再继续货架空了前台必须停下来等后厨烤出新面包再叫卖这就是同步问题。于是线程互斥解决“能不能挤进去”的问题线程同步解决“该等就等、该叫就叫”的问题。生产消费模型之所以被反复拿来当教学案例就是因为它把这两个问题同时暴露在了一个场景里既要保证多个线程不会同时改缓冲区又要保证缓冲区满和空的时候线程能正确等待和唤醒。把这一套代码写明白并发编程的地基也就打牢了。2. 线程同步与互斥两个容易混淆的概念2.1 临界区与竞态条件先把概念拉齐我发现很多人把“同步”和“互斥”当成一回事其实它们的侧重点完全不同。互斥重点在“排他”同一时刻只允许一个线程进入临界区同步重点在“协作”保证多个线程按照合理的顺序推进该等的等该继续的继续。打个比方只有一个座位的洗手间门口挂了个牌子谁拿牌子谁进去其他人只能在外面等。这个“牌子”就是互斥锁。但光有牌子还不够排队的人不知道里面的人什么时候出来只能干等。这时候需要一套叫号机制里面的人出来时按一下铃外面排队的人听到铃响就准备进。这个铃就是条件变量。互斥解决的是空间上的冲突条件变量解决的是时间上的等待。从代码角度看临界区不一定要很大可能只有一句count也可能是一整段业务逻辑。关键是所有线程访问共享数据的那条路径都得经过同一把锁。你要是只给其中一个生产者加锁另一个生产者不按规则来那锁跟没加一样。这是个很容易被忽略的认知锁保护的不是数据本身而是“访问数据的代码路径”。2.2 互斥锁给临界区上一道门在 Linux 的 C 语言多线程编程里互斥锁就是pthread_mutex_t使用起来非常简单pthread_mutex_t mutex PTHREAD_MUTEX_INITIALIZER; pthread_mutex_lock(mutex); // 临界区访问共享资源 pthread_mutex_unlock(mutex);但简单归简单使用时有几个坑是新手必踩的加锁范围太小只锁了写操作没锁读操作或者只锁了其中一段结果另一段代码照样修改共享变量竞态仍然存在。加锁范围太大把无关的耗时逻辑也锁在里面其他线程全部排队等待性能直线下降。忘记解锁函数中间有个return锁没释放其他线程永久阻塞。多把锁顺序不一致线程 A 持有锁 1 再去拿锁 2线程 B 持有锁 2 再去拿锁 1两边互相等对方释放直接死锁。这里有一个我后来养成的习惯持锁时间尽量短。临界区内只做必须受保护的“读改写”操作不要把printf、网络请求、文件写入这种慢操作放进锁里。很多人会顺手在临界区里打日志日志系统本身可能也要加锁锁套锁轻则性能崩重则死锁。先遍历一遍代码把所有共享变量的访问都找出来再决定锁的范围比写完了再调试省心得多。2.3 条件变量让线程学会等待和通知互斥锁只能保证同一时刻一个人进厨房但它解决不了“货架满了生产者为什么要停下来”这个问题。你总不能加个循环让生产者不停地尝试往里面塞塞不进去就继续塞那叫忙等。忙等会把 CPU 吃满而且由于线程调度不确定可能一直轮不到消费者执行生产者也一直塞不进去整个程序虽然没死但性能已经废了。条件变量pthread_cond_t就是用来做“等待与通知”的。关键 API 有三个函数作用pthread_cond_wait(cond, mutex)原子地释放 mutex并阻塞当前线程被唤醒后重新获取 mutex 再返回pthread_cond_signal(cond)唤醒一个正在等待该条件变量的线程pthread_cond_broadcast(cond)唤醒所有正在等待该条件变量的线程pthread_cond_wait的设计很奇怪它要求你先持有 mutex再调用它然后在内部把 mutex 释放掉。为什么因为条件判断和等待这两个动作必须是原子的。你总不希望线程 A 刚判断完“count 不为 0可以消费”正准备取数据的时候线程 B 把 count 改成了 0那 A 就拿错数据了。于是标准写法就固定成了这样pthread_mutex_lock(mutex); while (!condition) { pthread_cond_wait(cond, mutex); } // 条件满足执行临界区操作 pthread_mutex_unlock(mutex);注意这里用的是while而不是if这个细节我后面会专门展开讲。现在先记着条件变量负责让人睡下去、被人叫醒互斥锁负责保护整个判断和操作的过程两者永远是配合使用的缺一不可。3. 从零写一个生产消费模型3.1 设计选择环形队列和两个条件变量代码层面要实现一个生产消费模型最简单的设计是用固定大小数组做环形队列。环形队列的好处是空间复用in指针指向下一个写入位置out指针指向下一个读出位置超出数组末尾就取模回到头部。配合一个count字段记录当前积压数量不需要额外记录队列里每个位置是否有数据。我选择用两个条件变量而不是一个。not_full用来通知“队列有空位了”生产者等在它上面not_empty用来通知“队列有数据了”消费者等在它上面。如果只用一个条件变量队列满的时候生产者等待队列空的时候消费者等待生产者生产一个后会唤醒等待线程但被唤醒的可能是另一个生产者它发现队列还是满的只能继续睡白白浪费一次唤醒和锁竞争。两个条件变量可以把“满”和“空”这两类等待拆开唤醒更精准。数据结构长这样#define BUFFER_SIZE 5 typedef struct { int buffer[BUFFER_SIZE]; int in; // 下一个写入位置 int out; // 下一个读取位置 int count; // 当前积压的数据数量 pthread_mutex_t mutex; pthread_cond_t not_full; pthread_cond_t not_empty; } bounded_buffer_t;这种结构就是经典的“有界缓冲区”一个互斥锁保护所有字段和数组下标两个条件变量分别管两个方向的等待。逻辑不复杂但它把所有并发要点都覆盖了。3.2 完整代码可直接编译运行我用 C 和 pthread 写一版完整可运行的生产消费模型。两个生产者线程两个消费者线程生产者分别生成1xx和2xx编号的数据方便区分。代码量不大建议你自己敲一遍别直接复制粘贴因为手敲的过程中你会开始思考每行代码的顺序问题。#include stdio.h #include stdlib.h #include pthread.h #include unistd.h #include time.h #define BUFFER_SIZE 5 typedef struct { int buffer[BUFFER_SIZE]; int in; int out; int count; pthread_mutex_t mutex; pthread_cond_t not_full; pthread_cond_t not_empty; } bounded_buffer_t; void bb_init(bounded_buffer_t *bb) { bb-in 0; bb-out 0; bb-count 0; pthread_mutex_init(bb-mutex, NULL); pthread_cond_init(bb-not_full, NULL); pthread_cond_init(bb-not_empty, NULL); } void bb_produce(bounded_buffer_t *bb, int item) { pthread_mutex_lock(bb-mutex); while (bb-count BUFFER_SIZE) { pthread_cond_wait(bb-not_full, bb-mutex); } bb-buffer[bb-in] item; bb-in (bb-in 1) % BUFFER_SIZE; bb-count; printf([生产] item%d count%d\n, item, bb-count); pthread_cond_signal(bb-not_empty); pthread_mutex_unlock(bb-mutex); } int bb_consume(bounded_buffer_t *bb) { pthread_mutex_lock(bb-mutex); while (bb-count 0) { pthread_cond_wait(bb-not_empty, bb-mutex); } int item bb-buffer[bb-out]; bb-out (bb-out 1) % BUFFER_SIZE; bb-count--; printf([消费] item%d count%d\n, item, bb-count); pthread_cond_signal(bb-not_full); pthread_mutex_unlock(bb-mutex); return item; } typedef struct { bounded_buffer_t *bb; int id; } producer_arg_t; void *producer(void *arg) { producer_arg_t *pa (producer_arg_t *)arg; for (int i 1; i 20; i) { bb_produce(pa-bb, pa-id * 100 i); usleep(rand() % 500); } return NULL; } void *consumer(void *arg) { bounded_buffer_t *bb (bounded_buffer_t *)arg; for (int i 1; i 20; i) { int item bb_consume(bb); usleep(rand() % 500); } return NULL; } int main(void) { srand((unsigned)time(NULL)); bounded_buffer_t bb; bb_init(bb); producer_arg_t pa1 { bb, 1 }; producer_arg_t pa2 { bb, 2 }; pthread_t t1, t2, t3, t4; pthread_create(t1, NULL, producer, pa1); pthread_create(t2, NULL, producer, pa2); pthread_create(t3, NULL, consumer, bb); pthread_create(t4, NULL, consumer, bb); pthread_join(t1, NULL); pthread_join(t2, NULL); pthread_join(t3, NULL); pthread_join(t4, NULL); return 0; }编译命令很简单gcc -o pc pc.c -lpthread在 Linux 环境下直接运行即可。usleep(rand() % 500)是为了模拟生产者和消费者的处理耗时让线程调度更随机多跑几次更能体现出线程间的交错。3.3 关键点拆解为什么先加锁再判断这段代码里最值得弄清楚的是生产函数的开头三行pthread_mutex_lock(bb-mutex); while (bb-count BUFFER_SIZE) { pthread_cond_wait(bb-not_full, bb-mutex); }很多人第一次看这段会疑惑为什么已经加锁了还要在循环里再确认一次条件这个问题的答案恰恰是理解条件变量的关键。pthread_cond_wait被调用后会先释放 mutex再挂起线程。这意味着从你加锁到这行代码执行之间锁其实是“断开的”。如果这里用的是if当一个生产者被唤醒时它觉得自己可以生产了但唤醒它的可能不是消费者而是另一个线程的广播信号队列实际上还是满的。结果它继续执行往满队列里写数据覆盖掉还没被消费的数据。用while再检查一遍如果发现队列仍然满就继续睡直到真正有空位为止。同理消费者那边的判断也是while (bb-count 0)。这就是竞态条件下“唤醒后必须重查条件”的规范写法。条件变量不保存信号唤醒只是给你一个“去重新检查条件”的机会并不保证条件一定满足。还有一个顺序问题值得注意在修改完count之后我是先pthread_cond_signal再pthread_mutex_unlock。这两个操作的顺序在多数实现里都能工作但把 signal 放在 unlock 之前能让被唤醒的线程第一时间参与锁竞争减少等待线程“醒了却发现还没拿到锁”的间隙。严谨起见我会一直保持“先改共享状态再 signal最后 unlock”的习惯这个顺序也更容易在代码评审时说清楚。3.4 一个误区只加锁不引入条件变量行不行有些人可能会想既然是互斥锁保护队列那我每次操作前加锁、操作完解锁生产者放不进去就循环再试不引入条件变量行不行行但很难看。假设消费者线程改成这样pthread_mutex_lock(bb-mutex); while (bb-count 0) { pthread_mutex_unlock(bb-mutex); usleep(1000); pthread_mutex_lock(bb-mutex); }这其实就是在轮询。它也能让程序不崩溃但有两个问题一是消费有延迟明明数据刚来了消费者却还在 sleep等它醒来拿到数据已经是微秒甚至毫秒级之后二是频繁地拿锁、放锁、sleep、再拿锁CPU 空转不少线程多了以后锁竞争也更严重。条件变量存在的意义就是让消费者真正睡下去不占 CPU数据来了才被叫醒睡和醒都是事件驱动的效率高一个量级。4. 运行验证与测试场景4.1 单生产者单消费者最直观的节奏先把上面的代码简化成 1 个生产者、1 个消费者各跑 20 次。你会看到 buffer 里的count一会儿涨一会儿跌但始终保持在 0 到 5 之间[生产] item101 count1 [生产] item102 count2 [生产] item103 count3 [消费] item101 count2 [生产] item104 count3 [消费] item102 count2 ... [消费] item120 count0跑的过程中可以自己观察状态变化当count变成 5 的时候生产者会阻塞直到消费者取走一个数据当count变成 0 的时候消费者会阻塞直到生产者放入数据。这两个方向上的“等”就是条件变量在起作用。我建议你多跑几次每次结果都不太一样因为usleep引入了随机调度。但有一个规律是稳定的任何时候count都不会超过 5也不会小于 0。这说明互斥和同步都做对了。4.2 多生产者多消费者同一把锁下的秩序再用最开始的完整代码测试多生产者多消费者。两个生产者生产的编号分别是101~120和201~220两个消费者各消费 20 次。输出里会看到不同编号交错出现[生产] item101 count1 [生产] item201 count2 [消费] item101 count1 [生产] item102 count2 [生产] item202 count3 ...这种交错恰恰说明线程在并行跑但队列的数据没有被破坏。想进一步验证数据有没有丢或者重复可以把输出重定向到文件然后分别统计./pc out.txt grep 生产 out.txt | sort -n produced.txt grep 消费 out.txt | awk {print $2} | sort -n consumed.txt diff produced.txt consumed.txt理论上两个文件应该一模一样每一个被生产出来的 item 都恰好被消费一次。这能直观地证明虽然线程之间随机交错但共享队列的一致性没有被破坏。4.3 把锁去掉亲自看看竞态条件长什么样我建议你做一次“破坏性实验”把bb_produce里的pthread_mutex_lock和pthread_mutex_unlock注释掉再跑一次。大概率你会发现count会偶尔大于 5甚至消费者打印出count为负数。为什么是偶尔因为竞态条件本来就依赖特定的时序交错多线程调度的不确定性导致它不会每次都触发这也是并发 Bug 难排查的原因。这个实验不会影响生产环境只是让你亲眼看到没有互斥锁光靠“逻辑上正确”是不够的。CPU 的指令调度不会给你保证任何顺序你必须用同步原语把顺序显式地定下来。我教过不少人这一步“亲眼看到坏数据”比讲十遍理论都管用。5. 常见问题与排查实录5.1 死锁最常踩的三种姿势写生产消费模型最容易踩的第一个坑是死锁。最常见的情况是线程拿着锁去等待条件而另一个线程想修改条件却进不了临界区。比如有人会把代码写成这样pthread_mutex_lock(bb-mutex); while (bb-count 0) { pthread_mutex_unlock(bb-mutex); // 手动解锁后等待 pthread_cond_wait(bb-not_empty, bb-mutex); }手动在 wait 之前 unlock这是错误的。因为 unlock 和 wait 之间有一个窗口另一个线程可能在这期间消费掉最后一个数据然后发 signal但这个信号被当前线程错过了。更稳妥的做法就是标准语义调用pthread_cond_wait时必须已经持有锁wait 内部自己会释放锁。不要画蛇添足。第二种死锁是多把锁顺序不一致。如果生产逻辑里需要同时持有队列锁和日志锁消费者也持有这两个锁但加锁顺序反了就可能互相等待。解决办法很简单全局统一加锁顺序比如规定“永远是先队列锁再日志锁”。第三种情况更隐蔽函数中间return前忘记解锁。比如临界区里检查到某个异常状态直接return锁就没还回去。所以我现在写带锁的代码会习惯性地把临界区封装成独立函数保证每个入口和出口都成对减少忘记解锁的概率。排查死锁用 gdb 最有效。先找到经常卡住的进程号然后gdb -p pid进入 gdb 之后(gdb) info threads (gdb) thread apply all bt这条命令会打印出所有线程的调用栈一看就知道谁卡在pthread_cond_wait上谁卡在pthread_mutex_lock上死锁的锁依赖链基本一眼就能看穿。5.2 虚假唤醒为什么标准写法用 while 不用 ifLinux 的futex实现里条件变量的等待一般不会无缘无故被唤醒但 POSIX 标准明确允许pthread_cond_wait在没有 signal 的情况下返回这种情况叫虚假唤醒。再加上前面讲的“信号丢失 重新竞争锁”这个误差就会被放大。所以标准写法一定是while (条件不满足) { pthread_cond_wait(cond, mutex); }而不是if (条件不满足) { pthread_cond_wait(cond, mutex); }如果用if万一被虚假唤醒线程会直接跳过条件检查去操作共享数据轻则读到脏数据重则直接破坏队列结构。这个习惯要从写第一行并发代码就开始养成不要以为自己遇不到。还有一个点容易混淆pthread_cond_signal唤醒的是正在等待的线程但如果此时有多个线程同时等唤醒哪个是不确定的。要是你希望所有线程都尝试一下就要用pthread_cond_broadcast。对于更复杂的场景broadcast结合while重查条件才是更稳的组合。5.3 信号丢失条件变量的一个隐藏坑条件变量不像互斥锁那样会保存“状态”它里面没有记忆。如果在线程真正进入wait之前signal 就已经发出去了那么这个信号就丢了等待线程永远不会被唤醒。为了避免这个问题生产消费模型里一定要有“可检查的状态”。在我们的代码里这个状态就是count。消费者进入wait之前已经检查过count 0而生产者修改count之后才发信号顺序固定就不会出现“条件已经满足、但消费者还在睡”的极端情况。反过来就容易出事。假如你先发信号再修改count字段一个消费者被唤醒后立刻去检查count发现还是 0只好继续睡但此刻生产者还没来得及把 count 改成 1于是真正的“队列有数据”这个信号就丢了。所以我把“先改状态再 signal”当成一个铁律。5.4 实用调试命令与工具除了 gdb 之外有两个工具我强烈建议你装上。一个是 Valgrind 的 Helgrind专门检测多线程数据竞争valgrind --toolhelgrind ./pcHelgrind 能找到“多个线程访问同一块内存但没有加锁”的问题并且会给出访问栈。缺点是跑起来比较慢适合小规模测试。另一个是 ThreadSanitizer编译时直接插入检测代码gcc -g -fsanitizethread -o pc pc.c -lpthread ./pcTSan 运行时会输出所有数据竞争的详细信息包括冲突的线程 ID 和代码行号个人体验是比 Helgrind 更直接。唯一要注意的是加了-fsanitizethread之后程序运行会变慢且不能和 gdb 同时使用所以它是“检测用”的专属编译方式。5.5 性能优化方向从一个锁到多阶段队列上面的代码能跑对但并发度其实不高。因为一把互斥锁把所有生产者和消费者都串行化了同一时刻只能有一个线程操作队列其他线程都在等锁。当生产者消费者数量增加锁竞争就会成为瓶颈。想提升吞吐有几个方向可以尝试。第一个方向是分段锁。把一个大数组拆成多个小队列每个小队列有独立的锁生产者按编号散列到不同队列消费者也从对应队列取数。锁粒度变小了竞争自然减少。第二个方向是批量处理。消费者一次从队列里取出一批数据比如取 10 条再集中处理而不是取一条处理一条。这样可以减少加锁次数吞吐量提升非常明显。第三个方向是“单生产者单消费者模型”。如果你能保证只有一个生产者和一个消费者其实可以绕开锁用无锁的环形队列配合内存屏障实现。这是无锁编程里最经典也最安全的一种形态。真到了多生产多消费还要无锁那就要面对 CAS、ABA、内存序这些硬核问题复杂度比带锁方案高不少。我个人建议先把带锁版本跑熟再谈优化。多数业务场景下一把锁加合理的业务逻辑已经够用盲目上无锁队列只会让后续维护成本成倍增长。最后分享一点我的个人体会。写并发程序我现在的习惯是先画状态图把“什么条件下等、什么条件下唤醒、谁唤醒谁”画清楚再动手写代码。生产消费模型作为一个经典引例它的价值不只是教会你怎么用pthread_mutex_t和pthread_cond_t更是帮你建立一套思维方式多线程的世界里没有“理所当然”的顺序只有你亲手用同步原语定下来的秩序。把锁去掉跑一遍亲眼看到数据错乱的那一刻你才真正理解互斥和同步为什么必不可少。