1. 一条计数器的崩溃现场竞态条件到底怎么回事上一周我在调一个批量图片压缩工具开了四个线程同时去处理任务队列结果跑出来的图片里有好几张是花的还有一次直接段错误。我排查了很久最后定位到问题根源不在压缩算法而在最基础的地方——线程同步没有处理好。这个经历让我决定把Linux下的线程同步机制和生产者消费者模型彻底理一遍写成一篇能实际落地的文章。先解释一下为什么会数据错乱。多线程共享同一块内存时如果两个线程同时对同一个变量做写操作就会产生竞态条件race condition。比如一个简单的count在汇编层面不是一条指令而是读内存、加一、写回三步。两个线程同时走到第二步各自把旧值加一写回最终结果可能只加了两次中的一次甚至可能互相覆盖。这个场景听起来基础但真实项目里几十万行代码竞态条件往往就藏在某个你以为不会出问题的计数器里。下面我会分几块讲先讲清楚临界区是什么再用互斥锁解决互斥用条件变量解决等待然后给出生产者消费者模型的完整代码和信号量版本最后聊几个我在实测中翻过车的边界情况。1.1 从一次图片错乱开始我那个批量压缩工具的处理逻辑很简单主线程扫描目录生成任务列表工作线程从列表里取任务、压缩、写回磁盘。任务列表用的是一个动态数组为了省事我没有加锁当时想着取任务的时候只是读不会写应该安全吧。结果被直接打脸。两个工作线程可能会同时读到同一个尾部索引各自取到同一个任务ID于是同一个任务被压缩两次另一个任务却被漏掉。更隐蔽的是数组扩容的时候一个线程正在读另一个线程刚好realloc把内存搬走了轻则数据错乱重则直接段错误。这就是典型的读也要同步只要存在写操作任何并发访问都不能假定安全哪怕你只是读。从那以后我养成了一个习惯凡是共享可变数据一律先画出临界区再决定怎么保护。这也是我们接下来要做的第一件事——识别临界区。1.2 临界区、竞态条件和数据不一致的本质临界区critical section指的是访问共享资源的那段代码。线程同步本质上就是约定一种协议保证同一时刻只有一个线程进入临界区或者让多个线程按照某种顺序轮流进入。对临界区的保护要回答两个问题互斥Mutual Exclusion——同一时刻只能有一个线程在临界区内前进Progress——临界区外的线程不能无限阻塞其他线程。在实际编程里我们用互斥锁解决互斥用条件变量或信号量解决等待和唤醒这类协调问题。有些刚入门的同学会把原子操作和互斥锁混为一谈。原子操作适用于单变量、单指令级别的读写保护比如__atomic_add_fetch互斥锁适用于一块逻辑操作的串行化。拿厨房来类比原子操作是你端起一杯水互斥锁是给整个厨房装了一扇只能一个人进的门。你在里面炒菜的时候别人连切菜板都不能碰。2. 互斥锁给临界区装一扇只能单人通过的门互斥锁是Linux线程同步里最基础、也最容易被用错的工具。它的使用方式很简单进入临界区前加锁出临界区后解锁。但简单不等于不容易出问题锁的粒度和配对维护才是真正的难点。2.1 pthread_mutex 的基本姿势Linux下最常用的是POSIX线程库提供的pthread_mutex_t。初始化有两种方式静态初始化用宏PTHREAD_MUTEX_INITIALIZER动态初始化用pthread_mutex_init。日常写可复用的模块建议用动态初始化可以显式传入属性后面调属性也更方便。#include pthread.h pthread_mutex_t mutex PTHREAD_MUTEX_INITIALIZER; // 或者动态初始化 pthread_mutex_t mutex; pthread_mutex_init(mutex, NULL); // 使用 pthread_mutex_lock(mutex); // 临界区代码 pthread_mutex_unlock(mutex); // 销毁 pthread_mutex_destroy(mutex);这里我要强调一个很多新手容易忽略的点加锁和解锁必须严格配对。正常的函数返回路径要解锁出错提前返回的分支也要解锁。我在C语言项目里习惯用goto out_unlock的方式集中处理或者在函数开头就把锁的释放在所有出口落实一遍。C没有RAII不如C的std::lock_guard省心但集中释放比到处散落unlock更不容易漏。2.2 锁的粒度到底该选多细锁的粒度是线程同步里一个永远绕不开的取舍。锁太粗临界区执行时间长多线程退化成串行锁太细边界复杂很容易漏掉某个共享变量。举个例子一个共享任务队列如果只有push和pop内部各加一把锁看起来没问题但如果你在外面又要取队列长度又要判断是否为空用两个加锁操作之间的数据就可能在第二次加锁前被别人改掉。这时候要么把判断操作整体放进一个临界区要么单独用一把锁保护状态信息。我在上一节的图片压缩工具里最初就是只保护了push和pop没保护size字段结果消费者判断队列为空就不取生产者已经把size改了等消费者醒来时数据状态已经乱了。2.3 互斥锁解决不了等待只有互斥锁时消费者发现队列是空的只能循环重试。代码写出来像这样while (queue_empty(q)) { usleep(1000); } item queue_pop(q);这就是轮询。轮询有两个问题一是延迟不可控睡1000微秒可能刚好错过生产者放数据的时间点二是CPU浪费睡太短了空转睡长了延迟高。在嵌入式或者高并发服务里这种写法都非常难看。真正需要的是这样一个机制消费者明确地挂起生产者生产完数据后主动通知消费者来取。这就是条件变量派上用场的地方。3. 条件变量把轮流等消息变成来了我叫你条件变量本身不是锁它解决的问题是等让线程在某个条件不满足时睡眠等待另一个线程在条件满足后把它唤醒。它与互斥锁配合使用时能完美替代轮询。3.1 pthread_cond_wait 为什么必须配锁条件变量最核心的API是pthread_cond_wait(cond, mutex)这个函数的使用逻辑是调用方必须已经持有mutex也就是说你是在某个临界区里调用cond_wait的cond_wait内部会先释放mutex然后把当前线程挂起到cond的等待队列上被唤醒后它会重新尝试获取mutex拿到锁才返回。为什么必须这样设计因为要避免丢失唤醒。想象一个场景消费者检查到队列为空正准备睡眠此时生产者恰好放入一个数据并发送了唤醒信号。如果消费者的检查睡眠不是一个原子的过程那么唤醒信号可能在消费者真正睡眠之前就已经发出去了消费者醒不过来队列里的数据永远没人处理。cond_wait在内部把释放锁 挂起等待做成了原子操作检查过程和注册等待之间不再有缝隙生产者不可能在这段间隙里插入signal。我当初理解这个函数时把它想象成把门钥匙交给管理员、然后自己躺平睡觉管理员拿到钥匙开门放进东西再来拍醒我我醒来第一件事是拿回钥匙再进房间。这样就不会出现人还没睡货已经到了但没人喊醒我的尴尬。3.2 signal 和 broadcast 的差别pthread_cond_signal唤醒等待队列上的一个线程pthread_cond_broadcast唤醒全部。大多数单生产者单消费者场景用signal就够了。但有一个陷阱如果等待队列上有多个线程但它们等待的条件是同一个条件用signal没问题如果等待的条件不完全相同你又不知道唤醒的是哪一个就极可能唤醒一个刚醒来发现条件还是不满足的线程造成惊群和后续的无效竞争。这类场景直接用broadcast更稳妥。我在多消费者场景里遇到过一种情况两个消费者都在等队列非空生产者放了一个数据后用了signal理论上唤醒哪一个都行但如果其中一个消费者被其他线程持锁卡住了唤醒的线程拿不到锁也只能继续等另一个消费者却没有被通知到最终可能造成短暂的饥饿。虽然不会死锁但延迟变高了。后来我在代码里统一对队列状态发生变化的时刻用broadcast代价是可能多唤醒一个线程但逻辑更简单、更不容易出错。3.3 为什么判断条件要用 while 而不是 if条件变量的经典使用模式是pthread_mutex_lock(mutex); while (condition false) { pthread_cond_wait(cond, mutex); } // 这里是临界区操作 pthread_mutex_unlock(mutex);很多人会问既然cond_wait被唤醒时条件应当已经满足为什么还要while循环因为这世界比想象中更复杂。POSIX 标准允许条件变量存在虚假唤醒spurious wakeup即使没有线程发出signal等待的线程也可能被唤醒。另外即使没有虚假唤醒多个消费者同时等一个条件时一个broadcast会把所有线程都唤醒但它们会排队竞争锁等排到第二个线程时第一个线程可能已经把队列里的数据取光了条件再次变为不满足。所以while不是锦上添花而是必需。我在自己的代码里从不写if判断条件变量这条规则没有任何例外。4. 经典生产消费模型条件变量互斥锁的完整落地到这一步工具都齐了可以开始搭真正的生产者消费者模型。这个模型解决的问题是生产者往缓冲区里放数据消费者从缓冲区里取数据生产和消费的速度不一定匹配缓冲区必须既能存数据又不会在满时让生产者死等、在空时让消费者空转。4.1 共享缓冲区的数据结构设计我用的缓冲区是一个环形数组配合头尾指针和计数器。比起简单的链表环形数组的内存预先分配好没有频繁的malloc/free在吞吐量测试里表现稳定得多。#include stdio.h #include stdlib.h #include pthread.h #include unistd.h #define BUFFER_SIZE 8 typedef struct { int buf[BUFFER_SIZE]; int head; int tail; int count; pthread_mutex_t mutex; pthread_cond_t not_full; pthread_cond_t not_empty; } bounded_buffer_t; void buffer_init(bounded_buffer_t *bb) { bb-head 0; bb-tail 0; bb-count 0; pthread_mutex_init(bb-mutex, NULL); pthread_cond_init(bb-not_full, NULL); pthread_cond_init(bb-not_empty, NULL); }这里用两个条件变量not_full和not_empty分别等待缓冲区不满和缓冲区不空。这样可以做到生产者只被不满信号唤醒消费者只被不空信号唤醒互相之间不会误醒。4.2 生产者与消费者的核心操作生产者的核心逻辑是加锁如果缓冲区满就等not_full然后写入数据更新指针和计数发not_empty信号解锁。void buffer_push(bounded_buffer_t *bb, int val) { pthread_mutex_lock(bb-mutex); while (bb-count BUFFER_SIZE) { pthread_cond_wait(bb-not_full, bb-mutex); } bb-buf[bb-tail] val; bb-tail (bb-tail 1) % BUFFER_SIZE; bb-count; pthread_cond_signal(bb-not_empty); pthread_mutex_unlock(bb-mutex); } int buffer_pop(bounded_buffer_t *bb) { pthread_mutex_lock(bb-mutex); while (bb-count 0) { pthread_cond_wait(bb-not_empty, bb-mutex); } int val bb-buf[bb-head]; bb-head (bb-head 1) % BUFFER_SIZE; bb-count--; pthread_cond_signal(bb-not_full); pthread_mutex_unlock(bb-mutex); return val; }注意pthread_cond_signal放在解锁之前还是之后是个常见的争论点。在互斥锁保护下放在解锁前完全没问题而且代码更紧凑不会出现已解锁但还没发信号的中间状态。我在实践中一直这么做实测没有发现性能差异。4.3 线程主函数与完整验证接下来创建生产者和消费者线程各循环100次。void *producer_thread(void *arg) { bounded_buffer_t *bb (bounded_buffer_t *)arg; for (int i 0; i 100; i) { buffer_push(bb, i); printf([P] produce %d\n, i); usleep(rand() % 1000); } return NULL; } void *consumer_thread(void *arg) { bounded_buffer_t *bb (bounded_buffer_t *)arg; for (int i 0; i 100; i) { int val buffer_pop(bb); printf([C] consume %d\n, val); usleep(rand() % 1000); } return NULL; } int main() { bounded_buffer_t bb; buffer_init(bb); pthread_t p, c; pthread_create(p, NULL, producer_thread, bb); pthread_create(c, NULL, consumer_thread, bb); pthread_join(p, NULL); pthread_join(c, NULL); return 0; }编译时记得链接线程库gcc -g -O2 -o prod_cond prod_cond.c -lpthread ./prod_cond验证的重点不是看控制台打印是否整齐而是确认消费的数据集合和生产的数据集合完全一致。我把生产者产生的每个数字都记在数组里消费者消费后再比对跑了十轮数据不重不漏这个模型才算通过。如果只看日志很容易被交错的输出误导。5. 信号量版本两条信号线也能玩出同样的效果除了条件变量Linux线程同步里还有一个经典工具信号量semaphore。它本质是一个计数器加上等待队列sem_wait把计数器减一如果已经是0就阻塞sem_post把计数器加一并唤醒一个等待者。5.1 用两个信号量模拟缓冲区水位生产消费模型用信号量实现非常自然一个信号量empty表示空槽数量初始化为缓冲区大小一个信号量full表示已占用槽数量初始化为0。生产者在放入数据前先sem_wait(empty)消费者在取出数据前先sem_wait(full)放入和取出后分别sem_post(full)和sem_post(empty)。#include semaphore.h #define BUFFER_SIZE 8 sem_t empty_slots; sem_t full_slots; pthread_mutex_t mutex; void *producer_thread(void *arg) { for (int i 0; i 100; i) { sem_wait(empty_slots); pthread_mutex_lock(mutex); // 放入缓冲区 pthread_mutex_unlock(mutex); sem_post(full_slots); } return NULL; } void *consumer_thread(void *arg) { for (int i 0; i 100; i) { sem_wait(full_slots); pthread_mutex_lock(mutex); // 取出缓冲区 pthread_mutex_unlock(mutex); sem_post(empty_slots); } return NULL; } int main() { sem_init(empty_slots, 0, BUFFER_SIZE); sem_init(full_slots, 0, 0); pthread_mutex_init(mutex, NULL); // 创建线程、join等 return 0; }这里的pthread_mutex_lock只用来保护放入/取出缓冲区这段临界区不保护信号量本身因为sem_wait和sem_post内部是线程安全的。这个版本比起条件变量版本代码更短语义更接近有多少资源可用。5.2 条件变量方案和信号量方案的取舍两种方案都能实现生产消费模型选择哪一种取决于场景。我整理了一个对比表方便你做决定对比维度条件变量 互斥锁信号量代码直观度需要理解cond_wait的原子释放锁机制逻辑稍多语义直观计数器就是资源数量多条件等待可以用多个条件变量精细控制信号量本身只有资源有/无复杂协调要自己组合误用风险容易忘while、容易丢失唤醒容易忘记sem_destroy或信号量计数错乱跨进程使用pthread条件变量较麻烦有名信号量/无名信号量都支持跨进程适用场景复杂依赖关系的同步资源池限流、经典生产消费我个人写生产消费模型默认用条件变量版本因为后续如果要在等待条件上增加优先级或批量逻辑条件变量更好扩展。但如果只是要限制并发数、实现一个简单的资源池信号量直接顶上简单可靠。6. 边界情况实测丢失唤醒、虚假唤醒与死锁生产消费模型的骨架搭好之后真正的考验才开始。我测试期间至少碰到了四类问题每一个在单线程下都不会暴露只有压测或长时间运行时才会蹦出来。6.1 丢失唤醒是怎么发生的又怎么防前面说cond_wait把释放锁挂起做成原子操作可以防丢失但如果代码里在检查条件之前就已经把锁放掉了丢失唤醒照样会发生。比如有人这样写pthread_mutex_unlock(mutex); while (count 0) { ... } pthread_cond_wait(cond, mutex); // 错误写法在unlock和cond_wait之间的窗口里生产者拿到了锁、放了数据、发了信号然后消费者才进入wait信号已经错过了。这种错误在code review里特别容易漏掉因为单次运行可能碰巧不出问题。我的经验是把cond_wait的调用方式固定成模板永远不要在wait前手动解锁。6.2 虚假唤醒没有 signal 也被唤醒我第一次遇到虚假唤醒时代码里用的是if (count 0) pthread_cond_wait(...)结果跑了一段时间后消费者取出了几个看起来没被生产过的数据。我一度以为是生产者漏数据了后来查资料才确认是条件变量存在虚假唤醒的可能。从那以后我的所有条件变量代码强制用while。如果你维护的是别人写的代码看到if cond_wait的组合建议直接改成while这是零成本提升健壮性的修改。6.3 死锁的几种经典姿势线程同步里死锁的典型原因有三个加锁顺序不一致。两个线程分别持有锁A、锁B然后互相等待对方手里的锁。解决办法是全局约定加锁顺序比如先锁队列锁再锁状态锁所有线程都按这个顺序来。持锁调用阻塞函数。在临界区里调用read、recv、sem_wait这类可能长时间阻塞的函数会让其他线程等锁等很久叠加外部超时可能演变成死锁。生产消费模型的临界区应该只做内存操作绝不塞IO。忘记解锁。每个pthread_mutex_lock都必须有对应的pthread_mutex_unlock尤其是在错误处理分支里特别容易漏。我用过 valgrind 的 helgrind 工具来排查这类问题valgrind --toolhelgrind ./prod_cond它会把可能的数据竞争和锁序问题直接打印出来。另外用 GCC 的 ThreadSanitizer 也能在运行时检测竞态gcc -g -fsanitizethread -o prod_cond_tsan prod_cond.c -lpthread ./prod_cond_tsanTSan 在检测到竞态时会输出详细的调用栈比人工翻日志高效得多。我现在的习惯是每写一段多线程代码先开 TSan 跑一轮再上压力测试。7. 再往前走一步环形队列、多消费者与性能观察基础版的生产消费模型跑通之后项目里往往还要面对更真实的场景不止一个消费者、缓冲区需要高吞吐、生产者消费者速度严重不匹配。7.1 环形队列为什么是高性能的好选择环形队列的内存是连续的读写指针通过取模循环使用避免了频繁分配释放。上文的bounded_buffer_t本身就是一个环形队列。它还有一个额外的好处生产者和消费者分别维护tail和head只需要靠count判断满空在单生产单消费场景下甚至可以退化成无锁设计——只要保证各自的指针只被自己线程写另一线程只读。7.2 多消费者场景要注意什么多个消费者从同一个环形队列里取数据pop操作必须在同一把锁保护下完成否则两个消费者可能取到同一个元素。但这里有个性能优化点把加锁的粒度控制在pop内部不要扩大到消费数据的处理逻辑。比如消费者取到图片文件名之后压缩和解码这类耗时操作不应该在锁内执行否则一个消费者做重活会阻塞其余消费者取下一个任务。7.3 观察性能的几个小手段压测时我习惯用perf top看热点如果某个锁相关的函数出现在热点前列说明锁竞争严重。另一个办法是统计线程实际等待锁的时长在lock前后分别取时间戳累加到全局计数器。等待时间占比超过20%就说明锁竞争需要优化这时候可以考虑分段锁、无锁队列或者用原子操作替代部分锁。我实测过一个简单场景缓冲区从8改为128生产者每批数据耗时不变消费者吞吐量提升了将近三成原因就是消费者被not_empty唤醒后能连续多取几次而不是取一个就睡。所以缓冲区大小的设置不是越大越好但太小了绝对会导致频繁唤醒和抖动。具体的值还是要结合你们任务的实际体积来调。7.4 一个亲测有效的排障习惯最后分享一个我个人的习惯给每个线程的日志加上线程名字全局序号。比如producer-3、consumer-5每条日志前面再打一个seq%ld的全局自增序号。这样出现问题后从日志能直接还原出事件发生的先后顺序而不需要靠编译器续命。多线程排障最大的难点就是看不懂时序有了全局序号很多问题一眼就能看出来是先空后满还是唤醒顺序错了。生产消费模型是Linux并发编程里绕不过去的经典题但真正把它写对、跑稳靠的不只是背API而是对临界区、条件变量、锁粒度和各种边界情况的敬畏。希望这篇里记录的这些实测经验能帮你少走几个我走过的弯路。
