前阵子给一个基于 Linux 的仓库管理项目做并发改造压测时遇到一个特别典型的毛病库存明明只有 100 件模拟 50 个线程同时下单跑完居然变成负数。一开始我以为是下单逻辑写错了后来用 ThreadSanitizer 一查问题很清楚——多个线程同时在改同一个变量数据竞争。这篇文章就把我从踩坑、定位到用互斥锁和原子操作修复的完整过程整理出来顺便把 Linux 下验证数据竞争的常用手段也讲了。做 C/C 多线程的朋友应该都能用上尤其是刚接触并发、被各种疑难 bug 折磨得头疼的初学者。1. 数据竞争是怎么发生的1.1 从一次库存扣减事故说起我最初写的库存扣减逻辑基本长这样#include stdio.h #include pthread.h #include unistd.h #define THREAD_NUM 50 #define LOOP_NUM 100000 static int stock 100; void *seller(void *arg) { (void)arg; for (int i 0; i LOOP_NUM; i) { if (stock 0) { usleep(10); stock--; } } return NULL; } int main(void) { pthread_t tids[THREAD_NUM]; for (int i 0; i THREAD_NUM; i) { pthread_create(tids[i], NULL, seller, NULL); } for (int i 0; i THREAD_NUM; i) { pthread_join(tids[i], NULL); } printf(stock %d\n, stock); return 0; }这段代码在单线程下毫无问题最多扣到 0 就退出。但多线程一跑stock 直接变成负数最夸张一次跑到了 -400 多。为什么会这样因为if (stock 0)和stock--这两步中间插了一个usleep它把线程的执行间隔放大了导致大量线程同时看到 stock 还是正数然后一起执行扣减。就算没有usleepstock--本身也不是安全的因为它在 CPU 层面并不是一条指令。用生活里的例子类比三个人同时看到桌上还剩最后一个苹果每个人都觉得自己伸手去拿是合理的结果三个人都拿了一个不存在的苹果。这不是数学问题而是并发过程中大家拿到的判断结果已经过期了。只要多个线程对一个可写共享变量的访问没有同步就会出现这种数据竞争。1.2 数据竞争的标准定义和未定义行为意味着什么在 C11 / C11 标准里数据竞争有一个明确的定义两个或多个线程同时访问同一内存位置其中至少有一个访问是写操作并且这些访问之间没有建立 happen-before 关系。这句话拆开看有三个条件缺一不可同一内存位置比如同一个全局变量、同一个数组元素。至少一个写两个线程都只读同一个变量不会产生数据竞争。没有同步没有通过锁、原子操作等机制让访问排序。违反这个规则程序就进入未定义行为undefined behavior。很多人对未定义行为没有概念以为最多就是结果不对。实际上编译器拿到这种代码完全可能基于没有数据竞争这个假设做激进优化生成一些让你完全摸不着头脑的指令。也就是说数据竞争不仅会让结果错还可能让程序崩溃、卡死甚至表现成完全无关的诡异现象而且同样的代码换一个编译优化级别表现又不一致。这也是我反复强调的一点不要用volatile来解决并发问题。volatile只是告诉编译器这个变量可能被外部改变不要优化掉,它不会生成任何原子指令也不会对多线程访问做排序约束。用volatile修饰共享变量该数据竞争还是数据竞争甚至可能掩盖问题让 bug 更难以复现。1.3 从 CPU 指令层面理解一行代码不是原子的为什么stock--在 C 语言里只有一行却不是原子操作因为编译成汇编之后它通常会被拆成三条指令mov把 stock 的值从内存加载到寄存器sub在寄存器里减 1mov把新值写回内存。如果线程 A 执行到第 2 步线程 B 也开始执行第 1 步那么 B 加载到的值还是旧值。两个线程都基于同一个旧值计算写回时自然就覆盖了对方的更新。这就是典型的读-改-写竞争。CPU 级别的缓存L1/L2会让这个问题变得更复杂每个 CPU 核心可能有自己的缓存副本如果一个核心还没把新值同步到主存另一个核心读到的就是旧副本。最终表现就是扣减次数丢失库存变负数。理解这一点之后解决思路就清晰了要么在读-改-写这一步外面加一把锁让同一时间只有一个线程能进入要么让读-改-写变成一条不可被拆分的原子指令例如LOCK XADD。互斥锁和原子操作分别对应这两条路线。2. 互斥锁让共享区同一时间只进一个人2.1 pthread_mutex 基本用法一把锁守护一个库存Linux 下最常用的线程同步工具就是 POSIX 线程库提供的互斥锁pthread_mutex_t。其实用起来不复杂核心就四个函数初始化、加锁、解锁、销毁。#include pthread.h static pthread_mutex_t stock_lock PTHREAD_MUTEX_INITIALIZER; static int stock 100; void *seller(void *arg) { (void)arg; for (;;) { pthread_mutex_lock(stock_lock); if (stock 0) { pthread_mutex_unlock(stock_lock); break; } stock--; pthread_mutex_unlock(stock_lock); usleep(10); // 模拟后续业务耗时不放在锁内 } return NULL; }把if判断、扣减操作全部放进lock和unlock之间任何时刻只有一个线程能执行这段临界区代码另外的线程都在pthread_mutex_lock里等着。前面那个 usleep 放大竞争 的问题也顺手解决了因为模拟耗时被挪到了锁外库存扣减这个最关键的操作被保护住了。需要注意几点。第一加锁后要记得每个分支都能解锁尤其是break之前。我见过不少新手在continue、break、return里忘了解锁结果第二次循环直接死锁。第二用PTHREAD_MUTEX_INITIALIZER做静态初始化就够了只有在需要设置属性比如递归锁、多进程共享时才需要pthread_mutex_initpthread_mutex_destroy。第三大部分函数都有返回值生产代码里至少要对pthread_mutex_lock的返回值做检查别像我早期 demo 代码一样全靠自觉。2.2 锁粒度、锁顺序与死锁多把锁时的设计原则单个共享变量用一把锁逻辑简单。但真实项目里共享资源往往不止一个比如仓库系统里既要扣库存又要更新订单状态还要写日志。很多人图省事直接把所有操作包在同一个大锁里结果并发吞吐量断崖式下降。另一部分人为了性能搞了多把锁结果稍不小心就死锁。死锁最常见的场景是两个线程各自持有一把锁又在等对方手里的另一把锁。比如线程 A 持有库存锁等待订单锁线程 B 持有订单锁等待库存锁。两人谁也不放手程序卡死。解决办法有两个硬性原则所有线程按同一个全局顺序加锁。比如规定必须先锁库存再锁订单。这样 A 不会在持库存锁时申请订单锁等待 B因为 B 也按这个顺序必须先拿到库存锁才能拿订单锁也就不存在反向等待。实在无法统一顺序就用pthread_mutex_trylock或pthread_mutex_timedlock。拿不到锁就释放自己已有的锁退一步重试。这相当于用不等待来打破死锁的循环等待条件。锁粒度则需要根据业务权衡。以库存扣减为例如果临界区只是if stock--那粒度已经足够小。但如果把这个操作和一堆日志打印、网络上报写进同一临界区性能会非常难看因为其他线程全在锁外面干等。我的习惯是临界区只放必须保护的共享资源访问能把耗时操作放外面就放外面。像上面代码里把usleep挪出锁区就是这个思路。2.3 互斥锁的性能代价临界区越小越好也不一定有人可能觉得那把临界区拆得越小越好最好是每条共享变量访问都单独加锁。这个想法同样有问题。加锁和解锁本身是有开销的尤其当大量线程在同一个锁上竞争时内核线程会被挂起和唤醒开销远大于直接执行几条指令。如果临界区太小锁开销占大头如果临界区太大并发度又低。这是一个需要实测的权衡点。性能方面我在这个仓库项目里做过一个简单 benchmark同样是 100 个线程各自扣 10 万次库存不加锁版本跑得最快但结果错加互斥锁版本大约耗时 2.8 秒用合理粒度的锁版本能压到 2 秒左右。这个数据受硬件、线程数、临界区代码影响很大但能说明一件事锁竞争是真实存在的成本尤其是在高并发场景下。还有一点容易被忽略默认的pthread_mutex是不可重入的。也就是说同一个线程在持有锁的情况下再次pthread_mutex_lock同一把锁会直接死锁。如果确实需要重入初始化时设置PTHREAD_MUTEX_RECURSIVE属性。但递归锁通常是代码设计有问题的信号优先应该重构而不是开递归。3. 原子操作不需要加锁的轻量级并发3.1 从 C11 stdatomic 到 GCC __atomic怎么选互斥锁能解决数据竞争但锁这个字意味着同一时间只有一个线程能干活竞争激烈时效率不高。对简单的计数、状态标志这类场景Linux 下可以改用原子操作。原子操作依赖 CPU 提供的指令级别支持比如 x86 平台的LOCK XADD不会被任务切换打断也不需要操作系统调度器介入。C11 标准引入了stdatomic.h用起来非常直观。改造一下库存扣减#include stdatomic.h static atomic_int stock ATOMIC_VAR_INIT(100); void *seller(void *arg) { (void)arg; for (;;) { int expected atomic_load(stock); if (expected 0) break; if (atomic_compare_exchange_weak(stock, expected, expected - 1)) { usleep(10); } } return NULL; }这里用到了atomic_compare_exchange_weak也就是 CASCompare-And-Swap。它的语义是比较 stock 当前值是否等于 expected如果等于就把它改成 expected - 1返回成功如果不等于就把 expected 更新为 stock 当前值返回失败。失败后循环重试。这个模式比先判断再扣减安全得多因为判断和扣减被合并成了一个不可拆分的原子操作。usleep放在 CAS 成功之后这时候库存已经被扣掉了其他线程看到的是新值不会出现多扣。如果你在写 GNU 扩展环境下的代码也可以用 GCC 提供的__atomic内置函数__atomic_sub_fetch(stock, 1, __ATOMIC_SEQ_CST);它和 C11 原子操作底层等价。项目里如果有历史代码用的是老 GCC用__atomic兼容性好新项目我建议直接用 C11 标准接口可移植性更强。3.2 内存序relaxed、acquire、release 和 seq_cst 到底在说啥原子操作比锁轻量不代表没有门道。最绕的是内存序memory order。简单说编译器为了性能会重排指令CPU 也可能乱序执行原子操作本身保证不会被切开但它不保证它周围其他普通内存访问的顺序。内存序就是告诉编译器和 CPU你要什么样的可见性约束。C11 定义了四种常用内存序memory_order_relaxed: 最宽松只保证原子性不限制重排。适合做纯计数器比如统计在线人数、累计请求次数。memory_order_acquire: 用在读操作上保证这个读之后的普通读写不会被重排到这个读之前。memory_order_release: 用在写操作上保证这个写之前的普通读写不会被重排到这个写之后。memory_order_seq_cst: 顺序一致性最严格也是atomic_load、atomic_fetch_sub这类默认操作使用的模型。acquire 和 release 经常成对出现。典型的生产者消费者模式生产者先把数据写到一个普通数组里再用atomic_store_explicit(..., memory_order_release)发布一个数据已就绪标志消费者先用memory_order_acquire读这个标志标志确认后就认为生产者之前的普通写操作全部可见。这样做比全程seq_cst性能更好因为它允许周围读写更大程度乱序同时关键边界不乱。不过对绝大多数业务代码我建议先用默认的seq_cst。不要一上来就优化内存序复杂的内存序很容易写出只在特定硬件上能跑对的代码。等确认这里确实是性能瓶颈再结合 CPU 架构手册逐条分析不迟。3.3 CAS 循环与无锁队列原子操作不止于计数原子操作可以做一些比数字加减更复杂的事比如实现单生产者单消费者的无锁环形队列。无锁并不代表没有等待而是不用让线程进入内核睡眠如果队列满了或者空了调用方可以自旋重试或者直接返回错误。下面这个简化版代码在仓库项目的速率采集模块里用过#include stdatomic.h #include stddef.h #include sched.h #define RING_SIZE 1024 struct ring_buf { int data[RING_SIZE]; atomic_size_t head; atomic_size_t tail; }; void rb_init(struct ring_buf *rb) { atomic_init(rb-head, 0); atomic_init(rb-tail, 0); } int rb_enqueue(struct ring_buf *rb, int val) { size_t h atomic_load_explicit(rb-head, memory_order_relaxed); size_t nh (h 1) % RING_SIZE; if (nh atomic_load_explicit(rb-tail, memory_order_acquire)) { return -1; // 队列满 } rb-data[h] val; atomic_store_explicit(rb-head, nh, memory_order_release); return 0; } int rb_dequeue(struct ring_buf *rb, int *val) { size_t t atomic_load_explicit(rb-tail, memory_order_relaxed); if (t atomic_load_explicit(rb-head, memory_order_acquire)) { return -1; // 队列空 } *val rb-data[t]; size_t nt (t 1) % RING_SIZE; atomic_store_explicit(rb-tail, nt, memory_order_release); return 0; }核心思路是用head和tail两个原子变量分别表示生产者和消费者的位置。生产者先写data[h]再更新head并释放消费者先读head并获取确认head更新意味着data数据已经写完然后读取data[t]。这里release/acquire的作用就是防止消费者在生产者真正写完数据之前读到了旧的数组内容。需要注意的是这个简化实现只适合单生产者单消费者。多生产者多消费者场景下nh tail的判断和head更新之间仍然有竞争需要更复杂的 CAS 机制或者引入辅助队列。此外还有一个经典问题叫 ABACAS 循环中值从 A 变成 B又变回 A另一个线程会误以为没有人改过它。解决 ABA 的常见办法是给原子变量加一个递增计数器比如把 64 位数拆成高 32 位计数器、低 32 位索引或者使用带标记的指针。但这些都属于高难度玩法日常业务建议先老老实实用锁。4. Linux 下如何排查与验证数据竞争4.1 用 ThreadSanitizer 跑出数据竞争报告遇到那种偶尔出一次、单线程又复现不了的问题我最推荐直接用 ThreadSanitizerTSan。它是编译器内置的检测工具GCC 和 Clang 都支持不需要装额外的外部工具修改一下编译参数就行。gcc -g -fsanitizethread -pthread test.c -o test ./test还是前面那段有问题的库存代码加上这个参数编译运行TSan 会在控制台输出一个典型的报告WARNING: ThreadSanitizer: data race Read of size 4 at 0x... #0 seller test.c:8 Previous write of size 4 at 0x... #1 seller test.c:10它会明确告诉你哪里发生了读哪里发生了写以及对应的源文件行号。拿到这个输出再回去对照代码基本一眼就看出问题。修复之后重新编译运行TSan 不再报警才能算初步过关。实际使用中有几个坑TSan 和 AddressSanitizerASan不能混用加了-fsanitizethread就不要同时加-fsanitizeaddress。还有 TSan 在高并发下会有一些误报尤其是它自己实现模拟环境时对某些系统调用不友好遇到可疑报告可以先减小线程数验证。另外 TSan 会导致程序变慢很多这是正常的它要记录大量内存访问信息。4.2 从日志乱序到 gdb 现场实战排查流程不是所有数据竞争都能靠 TSan 一次抓出来尤其当问题依赖特定时序时。我习惯的排查流程分三步。第一步先用压力测试复现问题。写一个多线程压测程序线程数不要太多也不要太少50 到 200 之间通常最容易暴露问题。每个线程执行尽量多的循环次数比如 10 万次以上同时在线程函数里加一些适度的sleep或usleep来制造交错窗口。若问题稳定出现后面的定位就省力很多。第二步加日志观察。在共享变量的读和写前后打印线程 ID、当前值和修改后的值。日志里如果看到线程 A 读到 5线程 B 也读到 5基本就能确认竞争。但要注意日志本身也会影响时序可能加了日志就复现不了了。所以日志要能开关并且尽量只做辅助。第三步如果还定位不到用 gdb 挂到正在运行的进程上gdb -p pid (gdb) info threads (gdb) thread apply all btinfo threads能列出所有线程thread apply all bt能打印每个线程的调用栈。通过观察哪个线程卡在哪个函数有时能直接发现多线程同时进入同一个临界区的畸形状态。配合条件断点比如break seller if stock -1可以在库存变负的那一瞬间停住再用watch stock监控变量变化。gdb 没法直接帮你判断数据竞争但能帮你缩小范围最终结论还是要靠代码逻辑和 TSan 交叉验证。4.3 选型建议仓库场景下锁、原子、无锁队列怎么搭修完一轮并发问题之后我整理了一个简单的选型表分享出来供参考典型场景推荐方案理由临界区包含复杂判断、多个共享变量、磁盘IO互斥锁原子操作无法覆盖多步复合逻辑简单计数器、库存增减、状态标志原子操作C11 / __atomic开销低无需进入内核等待单生产者单消费者的数据管道原子操作 环形缓冲无锁延迟低代码可控多生产者多消费者高频操作无锁队列或分段锁需要仔细处理 ABA、内存序先做压测验证拿仓库系统里的实际模块做例子全局库存扣减我用的是原子操作的 CAS 循环几乎没性能压力订单状态流转涉及多个状态判断和数据库写我用互斥锁保护内存状态再异步刷到数据库速率采集和日志上报这类管道用无锁环形队列最合适生产者和消费者各自独立推进。这里有个很实在的建议能不用无锁队列就不用。无锁编程的调试成本很高ABA、ABA、内存序这些坑每一个都能让人排查好几天。真到了必须用无锁的时候严格收敛使用范围只允许在低频创建销毁、且生产者消费者模型明确的模块里出现代码里多写注释把内存序的设计依据写清楚。最后再分享一个经验在 Linux 上写多线程代码不要总觉得加锁是万能的也不要迷信无锁性能最牛。一个现代 CPU 上原子操作比互斥锁轻量得多但前提是竞争不激烈。竞争激烈时原子操作反复 CAS 失败比锁更容易浪费 CPU。最好的做法是先写正确的锁版本跑通后做 profile确认瓶颈确实在并发同步上再考虑原子化或无锁化。我在这个仓库项目里就是这样一步步把库存从负数修到稳定把耗时从 2.8 秒降到 1 秒出头的。先把工具用熟练再追求性能这个顺序千万别反过来。
前阵子给一个基于 Linux 的仓库管理项目做并发改造压测时遇到一个特别典型的毛病库存明明只有 100 件模拟 50 个线程同时下单跑完居然变成负数。一开始我以为是下单逻辑写错了后来用 ThreadSanitizer 一查问题很清楚——多个线程同时在改同一个变量数据竞争。这篇文章就把我从踩坑、定位到用互斥锁和原子操作修复的完整过程整理出来顺便把 Linux 下验证数据竞争的常用手段也讲了。做 C/C 多线程的朋友应该都能用上尤其是刚接触并发、被各种疑难 bug 折磨得头疼的初学者。1. 数据竞争是怎么发生的1.1 从一次库存扣减事故说起我最初写的库存扣减逻辑基本长这样#include stdio.h #include pthread.h #include unistd.h #define THREAD_NUM 50 #define LOOP_NUM 100000 static int stock 100; void *seller(void *arg) { (void)arg; for (int i 0; i LOOP_NUM; i) { if (stock 0) { usleep(10); stock--; } } return NULL; } int main(void) { pthread_t tids[THREAD_NUM]; for (int i 0; i THREAD_NUM; i) { pthread_create(tids[i], NULL, seller, NULL); } for (int i 0; i THREAD_NUM; i) { pthread_join(tids[i], NULL); } printf(stock %d\n, stock); return 0; }这段代码在单线程下毫无问题最多扣到 0 就退出。但多线程一跑stock 直接变成负数最夸张一次跑到了 -400 多。为什么会这样因为if (stock 0)和stock--这两步中间插了一个usleep它把线程的执行间隔放大了导致大量线程同时看到 stock 还是正数然后一起执行扣减。就算没有usleepstock--本身也不是安全的因为它在 CPU 层面并不是一条指令。用生活里的例子类比三个人同时看到桌上还剩最后一个苹果每个人都觉得自己伸手去拿是合理的结果三个人都拿了一个不存在的苹果。这不是数学问题而是并发过程中大家拿到的判断结果已经过期了。只要多个线程对一个可写共享变量的访问没有同步就会出现这种数据竞争。1.2 数据竞争的标准定义和未定义行为意味着什么在 C11 / C11 标准里数据竞争有一个明确的定义两个或多个线程同时访问同一内存位置其中至少有一个访问是写操作并且这些访问之间没有建立 happen-before 关系。这句话拆开看有三个条件缺一不可同一内存位置比如同一个全局变量、同一个数组元素。至少一个写两个线程都只读同一个变量不会产生数据竞争。没有同步没有通过锁、原子操作等机制让访问排序。违反这个规则程序就进入未定义行为undefined behavior。很多人对未定义行为没有概念以为最多就是结果不对。实际上编译器拿到这种代码完全可能基于没有数据竞争这个假设做激进优化生成一些让你完全摸不着头脑的指令。也就是说数据竞争不仅会让结果错还可能让程序崩溃、卡死甚至表现成完全无关的诡异现象而且同样的代码换一个编译优化级别表现又不一致。这也是我反复强调的一点不要用volatile来解决并发问题。volatile只是告诉编译器这个变量可能被外部改变不要优化掉,它不会生成任何原子指令也不会对多线程访问做排序约束。用volatile修饰共享变量该数据竞争还是数据竞争甚至可能掩盖问题让 bug 更难以复现。1.3 从 CPU 指令层面理解一行代码不是原子的为什么stock--在 C 语言里只有一行却不是原子操作因为编译成汇编之后它通常会被拆成三条指令mov把 stock 的值从内存加载到寄存器sub在寄存器里减 1mov把新值写回内存。如果线程 A 执行到第 2 步线程 B 也开始执行第 1 步那么 B 加载到的值还是旧值。两个线程都基于同一个旧值计算写回时自然就覆盖了对方的更新。这就是典型的读-改-写竞争。CPU 级别的缓存L1/L2会让这个问题变得更复杂每个 CPU 核心可能有自己的缓存副本如果一个核心还没把新值同步到主存另一个核心读到的就是旧副本。最终表现就是扣减次数丢失库存变负数。理解这一点之后解决思路就清晰了要么在读-改-写这一步外面加一把锁让同一时间只有一个线程能进入要么让读-改-写变成一条不可被拆分的原子指令例如LOCK XADD。互斥锁和原子操作分别对应这两条路线。2. 互斥锁让共享区同一时间只进一个人2.1 pthread_mutex 基本用法一把锁守护一个库存Linux 下最常用的线程同步工具就是 POSIX 线程库提供的互斥锁pthread_mutex_t。其实用起来不复杂核心就四个函数初始化、加锁、解锁、销毁。#include pthread.h static pthread_mutex_t stock_lock PTHREAD_MUTEX_INITIALIZER; static int stock 100; void *seller(void *arg) { (void)arg; for (;;) { pthread_mutex_lock(stock_lock); if (stock 0) { pthread_mutex_unlock(stock_lock); break; } stock--; pthread_mutex_unlock(stock_lock); usleep(10); // 模拟后续业务耗时不放在锁内 } return NULL; }把if判断、扣减操作全部放进lock和unlock之间任何时刻只有一个线程能执行这段临界区代码另外的线程都在pthread_mutex_lock里等着。前面那个 usleep 放大竞争 的问题也顺手解决了因为模拟耗时被挪到了锁外库存扣减这个最关键的操作被保护住了。需要注意几点。第一加锁后要记得每个分支都能解锁尤其是break之前。我见过不少新手在continue、break、return里忘了解锁结果第二次循环直接死锁。第二用PTHREAD_MUTEX_INITIALIZER做静态初始化就够了只有在需要设置属性比如递归锁、多进程共享时才需要pthread_mutex_initpthread_mutex_destroy。第三大部分函数都有返回值生产代码里至少要对pthread_mutex_lock的返回值做检查别像我早期 demo 代码一样全靠自觉。2.2 锁粒度、锁顺序与死锁多把锁时的设计原则单个共享变量用一把锁逻辑简单。但真实项目里共享资源往往不止一个比如仓库系统里既要扣库存又要更新订单状态还要写日志。很多人图省事直接把所有操作包在同一个大锁里结果并发吞吐量断崖式下降。另一部分人为了性能搞了多把锁结果稍不小心就死锁。死锁最常见的场景是两个线程各自持有一把锁又在等对方手里的另一把锁。比如线程 A 持有库存锁等待订单锁线程 B 持有订单锁等待库存锁。两人谁也不放手程序卡死。解决办法有两个硬性原则所有线程按同一个全局顺序加锁。比如规定必须先锁库存再锁订单。这样 A 不会在持库存锁时申请订单锁等待 B因为 B 也按这个顺序必须先拿到库存锁才能拿订单锁也就不存在反向等待。实在无法统一顺序就用pthread_mutex_trylock或pthread_mutex_timedlock。拿不到锁就释放自己已有的锁退一步重试。这相当于用不等待来打破死锁的循环等待条件。锁粒度则需要根据业务权衡。以库存扣减为例如果临界区只是if stock--那粒度已经足够小。但如果把这个操作和一堆日志打印、网络上报写进同一临界区性能会非常难看因为其他线程全在锁外面干等。我的习惯是临界区只放必须保护的共享资源访问能把耗时操作放外面就放外面。像上面代码里把usleep挪出锁区就是这个思路。2.3 互斥锁的性能代价临界区越小越好也不一定有人可能觉得那把临界区拆得越小越好最好是每条共享变量访问都单独加锁。这个想法同样有问题。加锁和解锁本身是有开销的尤其当大量线程在同一个锁上竞争时内核线程会被挂起和唤醒开销远大于直接执行几条指令。如果临界区太小锁开销占大头如果临界区太大并发度又低。这是一个需要实测的权衡点。性能方面我在这个仓库项目里做过一个简单 benchmark同样是 100 个线程各自扣 10 万次库存不加锁版本跑得最快但结果错加互斥锁版本大约耗时 2.8 秒用合理粒度的锁版本能压到 2 秒左右。这个数据受硬件、线程数、临界区代码影响很大但能说明一件事锁竞争是真实存在的成本尤其是在高并发场景下。还有一点容易被忽略默认的pthread_mutex是不可重入的。也就是说同一个线程在持有锁的情况下再次pthread_mutex_lock同一把锁会直接死锁。如果确实需要重入初始化时设置PTHREAD_MUTEX_RECURSIVE属性。但递归锁通常是代码设计有问题的信号优先应该重构而不是开递归。3. 原子操作不需要加锁的轻量级并发3.1 从 C11 stdatomic 到 GCC __atomic怎么选互斥锁能解决数据竞争但锁这个字意味着同一时间只有一个线程能干活竞争激烈时效率不高。对简单的计数、状态标志这类场景Linux 下可以改用原子操作。原子操作依赖 CPU 提供的指令级别支持比如 x86 平台的LOCK XADD不会被任务切换打断也不需要操作系统调度器介入。C11 标准引入了stdatomic.h用起来非常直观。改造一下库存扣减#include stdatomic.h static atomic_int stock ATOMIC_VAR_INIT(100); void *seller(void *arg) { (void)arg; for (;;) { int expected atomic_load(stock); if (expected 0) break; if (atomic_compare_exchange_weak(stock, expected, expected - 1)) { usleep(10); } } return NULL; }这里用到了atomic_compare_exchange_weak也就是 CASCompare-And-Swap。它的语义是比较 stock 当前值是否等于 expected如果等于就把它改成 expected - 1返回成功如果不等于就把 expected 更新为 stock 当前值返回失败。失败后循环重试。这个模式比先判断再扣减安全得多因为判断和扣减被合并成了一个不可拆分的原子操作。usleep放在 CAS 成功之后这时候库存已经被扣掉了其他线程看到的是新值不会出现多扣。如果你在写 GNU 扩展环境下的代码也可以用 GCC 提供的__atomic内置函数__atomic_sub_fetch(stock, 1, __ATOMIC_SEQ_CST);它和 C11 原子操作底层等价。项目里如果有历史代码用的是老 GCC用__atomic兼容性好新项目我建议直接用 C11 标准接口可移植性更强。3.2 内存序relaxed、acquire、release 和 seq_cst 到底在说啥原子操作比锁轻量不代表没有门道。最绕的是内存序memory order。简单说编译器为了性能会重排指令CPU 也可能乱序执行原子操作本身保证不会被切开但它不保证它周围其他普通内存访问的顺序。内存序就是告诉编译器和 CPU你要什么样的可见性约束。C11 定义了四种常用内存序memory_order_relaxed: 最宽松只保证原子性不限制重排。适合做纯计数器比如统计在线人数、累计请求次数。memory_order_acquire: 用在读操作上保证这个读之后的普通读写不会被重排到这个读之前。memory_order_release: 用在写操作上保证这个写之前的普通读写不会被重排到这个写之后。memory_order_seq_cst: 顺序一致性最严格也是atomic_load、atomic_fetch_sub这类默认操作使用的模型。acquire 和 release 经常成对出现。典型的生产者消费者模式生产者先把数据写到一个普通数组里再用atomic_store_explicit(..., memory_order_release)发布一个数据已就绪标志消费者先用memory_order_acquire读这个标志标志确认后就认为生产者之前的普通写操作全部可见。这样做比全程seq_cst性能更好因为它允许周围读写更大程度乱序同时关键边界不乱。不过对绝大多数业务代码我建议先用默认的seq_cst。不要一上来就优化内存序复杂的内存序很容易写出只在特定硬件上能跑对的代码。等确认这里确实是性能瓶颈再结合 CPU 架构手册逐条分析不迟。3.3 CAS 循环与无锁队列原子操作不止于计数原子操作可以做一些比数字加减更复杂的事比如实现单生产者单消费者的无锁环形队列。无锁并不代表没有等待而是不用让线程进入内核睡眠如果队列满了或者空了调用方可以自旋重试或者直接返回错误。下面这个简化版代码在仓库项目的速率采集模块里用过#include stdatomic.h #include stddef.h #include sched.h #define RING_SIZE 1024 struct ring_buf { int data[RING_SIZE]; atomic_size_t head; atomic_size_t tail; }; void rb_init(struct ring_buf *rb) { atomic_init(rb-head, 0); atomic_init(rb-tail, 0); } int rb_enqueue(struct ring_buf *rb, int val) { size_t h atomic_load_explicit(rb-head, memory_order_relaxed); size_t nh (h 1) % RING_SIZE; if (nh atomic_load_explicit(rb-tail, memory_order_acquire)) { return -1; // 队列满 } rb-data[h] val; atomic_store_explicit(rb-head, nh, memory_order_release); return 0; } int rb_dequeue(struct ring_buf *rb, int *val) { size_t t atomic_load_explicit(rb-tail, memory_order_relaxed); if (t atomic_load_explicit(rb-head, memory_order_acquire)) { return -1; // 队列空 } *val rb-data[t]; size_t nt (t 1) % RING_SIZE; atomic_store_explicit(rb-tail, nt, memory_order_release); return 0; }核心思路是用head和tail两个原子变量分别表示生产者和消费者的位置。生产者先写data[h]再更新head并释放消费者先读head并获取确认head更新意味着data数据已经写完然后读取data[t]。这里release/acquire的作用就是防止消费者在生产者真正写完数据之前读到了旧的数组内容。需要注意的是这个简化实现只适合单生产者单消费者。多生产者多消费者场景下nh tail的判断和head更新之间仍然有竞争需要更复杂的 CAS 机制或者引入辅助队列。此外还有一个经典问题叫 ABACAS 循环中值从 A 变成 B又变回 A另一个线程会误以为没有人改过它。解决 ABA 的常见办法是给原子变量加一个递增计数器比如把 64 位数拆成高 32 位计数器、低 32 位索引或者使用带标记的指针。但这些都属于高难度玩法日常业务建议先老老实实用锁。4. Linux 下如何排查与验证数据竞争4.1 用 ThreadSanitizer 跑出数据竞争报告遇到那种偶尔出一次、单线程又复现不了的问题我最推荐直接用 ThreadSanitizerTSan。它是编译器内置的检测工具GCC 和 Clang 都支持不需要装额外的外部工具修改一下编译参数就行。gcc -g -fsanitizethread -pthread test.c -o test ./test还是前面那段有问题的库存代码加上这个参数编译运行TSan 会在控制台输出一个典型的报告WARNING: ThreadSanitizer: data race Read of size 4 at 0x... #0 seller test.c:8 Previous write of size 4 at 0x... #1 seller test.c:10它会明确告诉你哪里发生了读哪里发生了写以及对应的源文件行号。拿到这个输出再回去对照代码基本一眼就看出问题。修复之后重新编译运行TSan 不再报警才能算初步过关。实际使用中有几个坑TSan 和 AddressSanitizerASan不能混用加了-fsanitizethread就不要同时加-fsanitizeaddress。还有 TSan 在高并发下会有一些误报尤其是它自己实现模拟环境时对某些系统调用不友好遇到可疑报告可以先减小线程数验证。另外 TSan 会导致程序变慢很多这是正常的它要记录大量内存访问信息。4.2 从日志乱序到 gdb 现场实战排查流程不是所有数据竞争都能靠 TSan 一次抓出来尤其当问题依赖特定时序时。我习惯的排查流程分三步。第一步先用压力测试复现问题。写一个多线程压测程序线程数不要太多也不要太少50 到 200 之间通常最容易暴露问题。每个线程执行尽量多的循环次数比如 10 万次以上同时在线程函数里加一些适度的sleep或usleep来制造交错窗口。若问题稳定出现后面的定位就省力很多。第二步加日志观察。在共享变量的读和写前后打印线程 ID、当前值和修改后的值。日志里如果看到线程 A 读到 5线程 B 也读到 5基本就能确认竞争。但要注意日志本身也会影响时序可能加了日志就复现不了了。所以日志要能开关并且尽量只做辅助。第三步如果还定位不到用 gdb 挂到正在运行的进程上gdb -p pid (gdb) info threads (gdb) thread apply all btinfo threads能列出所有线程thread apply all bt能打印每个线程的调用栈。通过观察哪个线程卡在哪个函数有时能直接发现多线程同时进入同一个临界区的畸形状态。配合条件断点比如break seller if stock -1可以在库存变负的那一瞬间停住再用watch stock监控变量变化。gdb 没法直接帮你判断数据竞争但能帮你缩小范围最终结论还是要靠代码逻辑和 TSan 交叉验证。4.3 选型建议仓库场景下锁、原子、无锁队列怎么搭修完一轮并发问题之后我整理了一个简单的选型表分享出来供参考典型场景推荐方案理由临界区包含复杂判断、多个共享变量、磁盘IO互斥锁原子操作无法覆盖多步复合逻辑简单计数器、库存增减、状态标志原子操作C11 / __atomic开销低无需进入内核等待单生产者单消费者的数据管道原子操作 环形缓冲无锁延迟低代码可控多生产者多消费者高频操作无锁队列或分段锁需要仔细处理 ABA、内存序先做压测验证拿仓库系统里的实际模块做例子全局库存扣减我用的是原子操作的 CAS 循环几乎没性能压力订单状态流转涉及多个状态判断和数据库写我用互斥锁保护内存状态再异步刷到数据库速率采集和日志上报这类管道用无锁环形队列最合适生产者和消费者各自独立推进。这里有个很实在的建议能不用无锁队列就不用。无锁编程的调试成本很高ABA、ABA、内存序这些坑每一个都能让人排查好几天。真到了必须用无锁的时候严格收敛使用范围只允许在低频创建销毁、且生产者消费者模型明确的模块里出现代码里多写注释把内存序的设计依据写清楚。最后再分享一个经验在 Linux 上写多线程代码不要总觉得加锁是万能的也不要迷信无锁性能最牛。一个现代 CPU 上原子操作比互斥锁轻量得多但前提是竞争不激烈。竞争激烈时原子操作反复 CAS 失败比锁更容易浪费 CPU。最好的做法是先写正确的锁版本跑通后做 profile确认瓶颈确实在并发同步上再考虑原子化或无锁化。我在这个仓库项目里就是这样一步步把库存从负数修到稳定把耗时从 2.8 秒降到 1 秒出头的。先把工具用熟练再追求性能这个顺序千万别反过来。
![写智能工程机械毕业论文，别只盯“排行榜”：选对 AI 搭档，从挖掘机液压故障诊断说起 [特殊字符]](http://pic.xiahunao.cn/yaotu/写智能工程机械毕业论文，别只盯“排行榜”：选对 AI 搭档，从挖掘机液压故障诊断说起 [特殊字符])
