多线程同步机制详解：互斥锁、条件变量与信号量实战

1. 从一次数据错乱说起为什么裸奔的多线程迟早要出事刚接触多线程那会儿我写过一个批量处理订单的小工具。逻辑特别简单开五个线程每个线程从共享的任务队列里取一条记录处理完把计数器加一。单线程跑的时候一切正常结果一上多线程计数器每次跑出来的结果都不一样有时候是 498有时候是 503明明总共就 500 条数据。更离谱的是日志里偶尔会出现两条线程处理同一条记录的情况。当时我盯着屏幕看了半天代码逻辑翻来覆去检查了好几遍就是找不到问题。后来一位做后端的老哥看了一眼说了句你那个计数器加一的操作是读-改-写三步中间随时可能被别的线程插一脚。这句话点醒了我——问题根本不在业务逻辑而在于多个线程同时访问共享资源时没有任何协调机制。这就是线程同步要解决的核心问题。所谓线程同步不是让线程排队一个个执行那是串行而是让多个线程在访问共享资源时能够按照某种规则协调彼此的行为保证数据的一致性和程序的正确性。它解决的场景非常具体多个线程读写同一块内存、同一个文件、同一个连接池、同一个缓存只要存在共享和并发就必然需要同步。这篇文章适合两类人看一类是刚学多线程、被各种锁和关键字绕晕的新手另一类是写过并发代码但总是凭感觉加锁、出了问题不知道怎么排查的开发者。我会把常见的几种同步方式从原理到实操讲透包括它们各自适合什么场景、有什么坑、怎么选。文中所有代码示例都是我自己跑过的参数和结论可以直接参考。在展开之前先把一个基础概念钉死竞态条件。它指的是程序的输出依赖于多个线程执行的相对时序。只要你的代码里存在先检查后执行或者读-改-写这类复合操作且操作对象是共享的就存在竞态条件。同步机制的本质就是把这些复合操作变成原子操作或者给它们加上访问秩序。2. 互斥锁最基础也最容易用错的同步手段2.1 互斥锁到底锁住了什么互斥锁Mutex是线程同步里最基础的工具它的语义很简单同一时刻只允许一个线程持有锁其他线程想拿锁就得等着。但很多人对锁住了什么这件事理解是模糊的。锁本身不锁任何数据它锁的是代码路径。你在一段代码前后加上加锁和解锁等于告诉系统这段代码同一时间只能有一个线程进来。用一个生活类比公司只有一间会议室门上有把锁钥匙只有一把。谁拿到钥匙谁就能进去开会其他人只能在门外等等里面的人出来把钥匙还回来。这里的会议室就是共享资源钥匙就是锁。关键在于所有想进会议室的人都必须遵守先拿钥匙再进门的规则只要有一个人不遵守整个秩序就崩了。对应到代码里问题往往出在有人不遵守规则。比如你在一处加了锁另一处忘了加那没加锁的那条路径就成了漏洞。这也是互斥锁最容易踩的坑锁的覆盖范围必须完整所有访问共享资源的路径都要被同一把锁保护。2.2 一个能复现问题的计数器示例先看不加锁的版本用 Python 演示因为它的线程模型直观容易观察现象import threading counter 0 def worker(): global counter for _ in range(100000): counter 1 threads [threading.Thread(targetworker) for _ in range(5)] for t in threads: t.start() for t in threads: t.join() print(counter)理论上 5 个线程各加 10 万次结果应该是 500000。但实际跑下来结果通常在 30 万到 45 万之间浮动。原因就是counter 1在底层被拆成了三步读取 counter 的值、加一、写回。两个线程可能同时读到同一个值各自加一后写回等于有一次加法被覆盖了。加上锁之后import threading counter 0 lock threading.Lock() def worker(): global counter for _ in range(100000): with lock: counter 1 threads [threading.Thread(targetworker) for _ in range(5)] for t in threads: t.start() for t in threads: t.join() print(counter)这次结果稳定是 500000。with lock这种写法会在进入代码块时自动加锁退出时自动解锁即使中间抛异常也能保证解锁比手动lock.acquire()和lock.release()安全得多。2.3 锁的粒度太粗拖慢性能太细容易漏锁的粒度指的是锁保护的代码范围有多大。粒度太粗比如把整个循环都包在锁里那多线程就退化成串行了性能还不如单线程。粒度太细比如只锁住写回那一步读和加一还在外面那锁就形同虚设。我一般的做法是只锁住真正访问共享资源的那几行。像上面计数器例子锁住counter 1就够了循环本身、函数调用这些不需要锁。判断标准很简单——问自己这几行代码有没有读写共享变量有就锁没有就不锁。但这里有个反直觉的点有时候锁的粒度不能只按共享变量来划。如果两个共享变量之间存在不变式关系比如账户 A 减一百、账户 B 加一百两者之和必须不变那这两个操作必须放在同一把锁里否则中间状态会被其他线程看到。这就是为什么转账操作必须整体加锁不能拆成两次独立的加锁。2.4 死锁两个线程互相等对方手里的锁死锁是多线程里最经典的问题。场景是这样的线程 1 拿到了锁 A正在等锁 B线程 2 拿到了锁 B正在等锁 A。两边都不肯放手程序就卡死了。import threading lock_a threading.Lock() lock_b threading.Lock() def task1(): with lock_a: print(task1 got lock_a) with lock_b: print(task1 got lock_b) def task2(): with lock_b: print(task2 got lock_b) with lock_a: print(task2 got lock_a) t1 threading.Thread(targettask1) t2 threading.Thread(targettask2) t1.start() t2.start() t1.join() t2.join()这段代码跑起来大概率会卡住。避免死锁有几个实用原则统一加锁顺序所有线程都按 A 再 B 的顺序拿锁、使用带超时的锁拿不到就放弃并释放已持有的锁、尽量减少同时持有的锁数量。我个人最推荐第一条因为它在设计阶段就能规避问题不需要运行时处理。注意死锁不一定发生在两把锁之间。一把锁也可能死锁比如同一个线程重复获取同一把不可重入的锁。这种叫自死锁排查时容易被忽略。3. 条件变量让线程学会等通知而不是空转3.1 忙等待为什么是浪费有些场景下线程需要等某个条件成立才能继续。比如消费者线程要等队列里有数据才能取生产者线程要等队列有空位才能放。最朴素的做法是写个循环不停检查while not queue: pass # 空转等待这种叫忙等待线程在疯狂消耗 CPU 做无用功。如果队列要等好几秒才有数据这几秒 CPU 就被白白烧掉了。更糟的是如果多个线程都在忙等CPU 占用率会飙到很高但实际什么活都没干。条件变量就是来解决这个问题的。它的思路是条件不满足时线程主动让出 CPU 进入等待状态当条件可能满足时由另一个线程发通知把它唤醒。这样等待期间线程不占 CPU被唤醒后才重新检查条件。3.2 条件变量必须和锁配合使用条件变量不能单独用它必须绑定一把互斥锁。原因在于检查条件和进入等待这两步之间必须原子否则会出现丢失唤醒的问题。想象一下线程 A 检查队列为空准备进入等待就在这个间隙线程 B 往队列里放了数据并发出了通知然后线程 A 才真正进入等待——结果就是 A 永远等不到那个已经错过的通知。标准用法是这样的import threading queue [] lock threading.Lock() cond threading.Condition(lock) def consumer(): with cond: while not queue: cond.wait() item queue.pop(0) print(fconsumed {item}) def producer(): with cond: queue.append(data) cond.notify()这里有几个细节值得说。第一cond.wait()会自动释放锁并进入等待被唤醒时会重新获取锁所以外面用with cond包着是安全的。第二条件检查必须用while而不是if因为被唤醒后条件不一定真的成立了可能被其他线程抢先消费了需要重新检查。第三notify()只唤醒一个等待线程notify_all()唤醒全部选哪个取决于你的场景。3.3 生产者消费者模型的完整实现把上面的片段补全就是一个经典的生产者消费者模型import threading import time import random queue [] MAX_SIZE 5 lock threading.Lock() cond threading.Condition(lock) def producer(pid): for i in range(10): with cond: while len(queue) MAX_SIZE: cond.wait() item fp{pid}-{i} queue.append(item) print(fproduced {item}, queue size {len(queue)}) cond.notify_all() time.sleep(random.uniform(0.1, 0.3)) def consumer(cid): while True: with cond: while not queue: cond.wait() item queue.pop(0) print(fconsumer {cid} consumed {item}, queue size {len(queue)}) cond.notify_all() time.sleep(random.uniform(0.1, 0.3)) producers [threading.Thread(targetproducer, args(i,)) for i in range(2)] consumers [threading.Thread(targetconsumer, args(i,), daemonTrue) for i in range(3)] for p in producers: p.start() for c in consumers: c.start() for p in producers: p.join()这个模型里生产者等队列不满消费者等队列不空两个条件共用一把锁和一个条件变量。用notify_all()是因为生产者和消费者等待的是不同条件只唤醒一个可能唤醒错对象导致该醒的没醒。3.4 条件变量和轮询的取舍条件变量不是万能的。如果条件变化非常频繁或者等待时间极短条件变量的加锁、等待、唤醒这一套开销可能比简单轮询还大。我一般这样判断等待时间可能超过毫秒级用条件变量等待时间在微秒级且变化频繁可以考虑自旋等待。但绝大多数业务场景下条件变量都是更优解因为业务里的等待往往是秒级甚至更长的。4. 信号量控制同时访问资源的线程数量4.1 信号量和互斥锁的本质区别互斥锁的值只有 0 和 1同一时刻只允许一个线程进入。信号量则是一个计数器可以设置成任意正整数 N表示同一时刻允许 N 个线程同时访问。这是两者最本质的区别。打个比方互斥锁是单人厕所一次只能进一个人信号量是停车场有 N 个车位前 N 辆车可以直接进第 N1 辆就得等有人开走才能进。信号量的核心操作有两个acquire申请一个许可计数器减一减到零就阻塞和release释放一个许可计数器加一唤醒等待的线程。4.2 用信号量限制并发连接数一个很实际的应用场景是限制对某个外部服务的并发请求数。假设某个接口最多承受 10 个并发连接超过就会拒绝服务那就可以用信号量把并发控制在 10 以内import threading import time semaphore threading.Semaphore(10) def call_service(task_id): with semaphore: print(ftask {task_id} acquired permit) time.sleep(1) # 模拟请求耗时 print(ftask {task_id} released permit) threads [threading.Thread(targetcall_service, args(i,)) for i in range(50)] for t in threads: t.start() for t in threads: t.join()50 个任务同时启动但任意时刻最多只有 10 个在执行其余的在with semaphore处排队。这种模式在爬虫限速、数据库连接池、API 调用限流里非常常见。4.3 信号量做互斥锁用可以但没必要信号量初始值设为 1 时行为上等价于互斥锁。但我不建议这么用原因有两个一是语义不清晰读代码的人看到信号量会以为你在控制并发数结果发现是 1容易困惑二是信号量没有持有者的概念任何线程都能 release而互斥锁通常要求谁加锁谁解锁这个约束在排查问题时很有价值。工具要用在对的语义上别为了省事混用。4.4 信号量的常见坑release 次数不匹配信号量最容易出的问题是 acquire 和 release 次数不匹配。多 release 一次计数器就虚高等于放宽了并发限制少 release 一次计数器就永久减少最终可能导致所有线程都卡住。所以用信号量时release 一定要放在 finally 块里或者用with语句确保异常情况下也能释放。# 不推荐 semaphore.acquire() do_something() # 这里抛异常release 就执行不到了 semaphore.release() # 推荐 with semaphore: do_something()5. 读写锁与原子操作读多写少场景的优化思路5.1 读写锁解决的是什么问题互斥锁有个明显的浪费如果多个线程都只是读共享数据彼此之间根本不会冲突但互斥锁还是让它们排队。读操作之间应该是可以并发的只有写操作才需要独占。读写锁就是为这个场景设计的。读写锁的规则是读锁可以被多个线程同时持有写锁只能被一个线程持有且写锁和读锁互斥。也就是说读读并发、读写互斥、写写互斥。对于读多写少的场景比如配置缓存、路由表、字典查询读写锁能显著提升吞吐量。不过要注意读写锁的实现复杂度比互斥锁高加解锁的开销也更大。如果临界区非常短或者读写比例没那么悬殊用读写锁反而可能更慢。我一般只在读操作明显多于写操作比如 10:1 以上且临界区有一定耗时时才考虑它。5.2 原子操作不需要锁的同步原子操作指的是不可被中断的一个或一系列操作它在硬件层面保证要么全部完成要么完全没执行中间不会被其他线程插入。最典型的就是原子自增、原子比较并交换CAS。用原子操作做计数器就不需要加锁了import threading counter 0 lock threading.Lock() # 加锁版本 def increment_with_lock(): global counter for _ in range(100000): with lock: counter 1 # 很多语言提供了原子类型比如 Java 的 AtomicInteger # Java 示例 # AtomicInteger counter new AtomicInteger(0); # counter.incrementAndGet();原子操作的优势是性能好因为它不需要操作系统介入做线程调度直接在 CPU 指令层面完成。但它的适用范围有限只能用于单个变量的简单操作。一旦涉及多个变量的复合逻辑还是得回到锁。5.3 CAS 的 ABA 问题CAS 操作有个著名的坑叫 ABA 问题。假设线程 1 读到变量值是 A准备把它改成 C在这期间线程 2 把 A 改成了 B又改回了 A线程 1 执行 CAS 时发现值还是 A就以为没人动过顺利改成 C。但实际上中间已经发生过变化如果这个变化对业务有影响就会出问题。解决 ABA 的常见办法是加版本号每次修改版本号加一CAS 时同时比较值和版本号。很多语言的标准库已经内置了带版本号的原子引用用的时候留意一下就行。5.4 各种同步方式的选型对照把前面讲的几种方式整理成一张表方便对照选择同步方式核心语义适用场景主要风险互斥锁同一时刻一个线程进入通用共享资源保护死锁、粒度不当条件变量等待条件成立再继续生产者消费者、等待通知丢失唤醒、虚假唤醒信号量限制同时访问的线程数限流、连接池、资源配额release 次数不匹配读写锁读并发、写独占读多写少的缓存场景写饥饿、开销偏大原子操作单变量不可中断操作计数器、标志位ABA 问题、适用范围窄选型的时候先问自己三个问题共享资源是什么类型读写比例大概多少临界区有多长答案基本就能定位到合适的工具。拿不准的时候先用互斥锁保证正确性等性能真的成为瓶颈再优化别一上来就追求花哨的方案。6. 排查同步问题的实战思路6.1 先确认问题是不是同步引起的不是所有多线程的诡异现象都是同步问题。有时候是逻辑 bug有时候是资源耗尽有时候是外部依赖不稳定。判断方法很简单把并发改成串行看问题是否消失。如果串行跑一百遍都正常并发跑几遍就出错那基本可以锁定是同步问题。另一个信号是结果不确定。同步问题导致的错误往往具有随机性同样的输入有时候对有时候错错误的位置也不固定。这种不确定性正是竞态条件的典型特征。6.2 缩小范围从共享变量入手定位同步问题时我会先把所有被多个线程访问的变量列出来然后逐个检查它们的访问路径。重点看三类操作读-改-写、先检查后执行、多个相关变量的联合更新。这三类是竞态条件的高发区。检查的时候有个技巧看有没有哪条访问路径没加锁。很多时候问题不是锁用错了而是某条路径漏加了锁。比如一个缓存读的时候加了锁写的时候忘了加那写操作就可能和读操作撞上。6.3 用日志和断点辅助定位同步问题难复现所以日志要打足。我一般会在加锁前后、条件等待前后、共享变量修改前后都打上带线程 ID 和时间戳的日志。这样一旦出问题能还原出各个线程的执行时序看出是谁在什么时候插了队。不过要注意打日志本身会影响时序有些问题加了日志就复现不出来了。这种情况可以用更轻量的方式比如记录关键操作的序号或者用专门的并发测试工具反复压测。6.4 几个我踩过的真实坑第一个坑是在锁里做耗时操作。有次我在锁保护的代码块里调了一个网络请求结果所有线程都堵在那把锁上吞吐量惨不忍睹。后来把网络请求挪到锁外面只在更新共享状态时加锁性能立刻上来了。教训是锁里只放必须原子执行的代码任何 IO、计算、等待都尽量挪出去。第二个坑是用可变对象做锁。有些语言里锁可以是任意对象但如果这个对象被重新赋值了锁就换了同步就失效了。所以锁对象一旦创建就不要改最好用专门的锁类型。第三个坑是忘记处理虚假唤醒。条件变量的 wait 可能在没有 notify 的情况下自己醒来虽然罕见但标准允许所以条件检查必须用 while 循环醒来后重新验证条件。用 if 的话虚假唤醒后会直接往下执行访问到还没准备好的数据。第四个坑是锁的嵌套顺序不一致。两个模块各自加锁A 模块先锁 X 再锁 YB 模块先锁 Y 再锁 X单独测都没问题一组合就死锁。这种问题在大型项目里特别隐蔽我的经验是给所有锁定一个全局顺序所有代码都按这个顺序加锁。7. 写在最后的一点个人体会线程同步这件事理论看着简单真写起来坑一个接一个。我自己的原则是能不用共享状态就不用。如果每个线程能有自己独立的数据通过消息传递来协作那大部分同步问题根本不会出现。这也是为什么很多现代并发模型比如 Actor 模型、协程加通道都在强调不要通过共享内存来通信而要通过通信来共享内存。但现实里完全避开共享状态不现实所以锁还是得会用。我的建议是先把互斥锁和条件变量这两个基础工具吃透它们能覆盖八成以上的场景。信号量、读写锁、原子操作这些等遇到具体需求再学别一上来就贪多。写并发代码的时候宁可保守一点多加点锁也别为了性能冒险省锁——正确性永远排在性能前面。最后分享一个我常用的自检清单写完任何一段并发代码我都会过一遍所有共享变量的访问路径都加锁了吗锁的粒度是不是刚好有没有可能死锁条件等待用的是 while 吗异常路径下锁能正常释放吗这几个问题都过了代码基本就稳了。