Python 死锁排查全攻略：从线程卡死到锁依赖定位与工程化修复-育师

Python 死锁排查全攻略：从线程卡死到锁依赖定位与工程化修复

在 Python 并发开发中，有一种故障比异常更让人头疼。

它没有Traceback，没有明显报错，CPU 可能也不高，日志甚至停留在一句看似正常的：

开始处理任务……

然后程序就再也没有然后了。

服务进程还活着，线程也还活着，请求却迟迟不返回；重启以后暂时恢复，运行一段时间又再次“假死”。

这类问题背后，一个非常典型的原因就是——死锁（Deadlock）。

Python 的threading模块让多线程开发非常方便，但多个线程共享内存，也意味着它们可能同时竞争Lock、RLock、Condition、Semaphore 等同步资源。当前 Python 官方文档明确说明：一个线程获取普通Lock后，其他再次获取该锁的操作会阻塞，直到锁被释放；acquire()也支持超时等待。(Python documentation)

真正危险的是：

如果持有锁的人，也正在等待另一个永远不会释放给它的锁，程序就可能永久僵住。

本文不只解释“什么叫死锁”，而是从 Python 实战角度完整回答：

Python 死锁是怎么产生的？
如何快速判断程序是慢还是死锁？
怎样抓取所有线程的调用栈？
如何根据线程栈还原锁依赖关系？
Lock和RLock到底怎么选？
如何通过锁顺序、超时、消息队列等方式从设计层消灭死锁？
线上程序偶发卡死时，应该按照什么顺序排查？

如果你正在学习 Python 并发编程，这是一份实战型 Python 教程；如果你已经维护生产系统，它也可以直接作为一份死锁排障清单。

一、什么是死锁？

先来看一个现实中的例子。

有两个人：

小王拿着会议室钥匙，等待投影仪。 小李拿着投影仪，等待会议室钥匙。

如果双方都不愿意先释放自己手中的资源：

小王 → 等小李 小李 → 等小王

两个人就会永远等待。

程序里的死锁完全类似。

假设：

Thread-A 已获得 lock_a Thread-B 已获得 lock_b

接下来：

Thread-A 等待 lock_b Thread-B 等待 lock_a

形成：

Thread-A │ ▼ lock_b ▲ │ Thread-B │ ▼ lock_a ▲ │ └──────── Thread-A

这就是一个循环等待链。

二、一个可以复现的 Python 死锁

看看下面的代码：

importthreadingimporttime lock_a=threading.Lock()lock_b=threading.Lock()defworker_a():print("A: 获取 lock_a")withlock_a:time.sleep(0.5)print("A: 等待 lock_b")withlock_b:print("A: 完成")defworker_b():print("B: 获取 lock_b")withlock_b:time.sleep(0.5)print("B: 等待 lock_a")withlock_a:print("B: 完成")t1=threading.Thread(target=worker_a,name="worker-a")t2=threading.Thread(target=worker_b,name="worker-b")t1.start()t2.start()t1.join()t2.join()print("程序结束")

运行以后，很可能看到：

A: 获取 lock_a B: 获取 lock_b A: 等待 lock_b B: 等待 lock_a

然后程序停止继续执行。

此时：

worker-a 持有 lock_a ↓ 等待 lock_b worker-b 持有 lock_b ↓ 等待 lock_a

双方都无法继续。

主线程又执行了：

t1.join()t2.join()

于是主线程也在等待它们退出。

最终整个程序看起来像“卡死”了一样。

Python 官方甚至专门禁止线程join()自己，因为这本身就会造成死锁，并会抛出RuntimeError。(Python documentation)

三、死锁成立通常需要什么条件？

理解死锁时，可以记住四个经典条件。

1. 互斥

某个资源同一时刻只能被一个线程持有。

例如：

lock=threading.Lock()

线程 A 获得锁后，线程 B 只能等待。

2. 持有并等待

线程拿着一个资源，同时等待另一个资源。

例如：

withlock_a:# 已经持有 Awithlock_b:...

如果拿不到lock_b，线程不会自动释放lock_a。

3. 资源不能被强制抢走

正常情况下，别人不能随意把线程正在使用的资源夺走。

因此等待线程只能继续等。

4. 循环等待

最终形成：

A 等 B B 等 C C 等 A

死锁真正麻烦的往往就是这个环。

所以排查死锁时，最关键的问题不是：

“哪个线程卡住了？”

而是：

“它在等待谁？那个线程又在等待谁？”

把等待关系连起来，问题通常就清楚了。

四、第一类常见死锁：锁顺序不一致

刚才的示例本质上就是：

线程 A： lock_a → lock_b 线程 B： lock_b → lock_a

解决办法非常经典：

统一锁获取顺序。

比如团队规定：

任何地方都必须： lock_a → lock_b

修改：

defworker_a():withlock_a:withlock_b:do_something()defworker_b():withlock_a:withlock_b:do_other_thing()

这样即使两个线程同时执行：

Thread-A → lock_a Thread-B → 等待 lock_a Thread-A → lock_b Thread-A → 完成并释放 Thread-B → 获得 lock_a

循环等待自然消失。

因此，多锁系统中最好建立明确的：

Lock Ordering

例如：

user_lock ↓ order_lock ↓ inventory_lock ↓ payment_lock

整个项目统一遵守。

这条规范往往比“出了问题再调试”便宜得多。

五、第二类死锁：自己把自己锁住

来看：

importthreading lock=threading.Lock()defouter():withlock:inner()definner():withlock:print("Hello")outer()

问题在哪？

outer()已经获取：

lock

随后调用inner()。

inner()又尝试：

withlock:

普通threading.Lock并不是可重入锁。

于是：

线程自己持有 lock ↓ 自己再次等待 lock ↓ 没人能够释放

形成自锁。

六、RLock 能解决什么？

如果业务设计确实需要同一个线程重复进入受保护区域，可以使用：

threading.RLock()

例如：

importthreading lock=threading.RLock()defouter():withlock:inner()definner():withlock:print("正常执行")outer()

RLock会记录：

锁的拥有线程 + 递归获取层数

同一个线程可以重复获取它，但获取多少次就必须相应释放多少次。官方文档因此建议优先通过with使用RLock，以减少获取、释放次数不匹配导致的问题。(Python documentation)

但是有一个很重要的误区：

RLock ≠ 万能死锁修复器

它主要解决：

同一个线程重复进入同一把锁

如果问题是：

线程 A：拿着 X 等 Y 线程 B：拿着 Y 等 X

换成RLock一样可能死锁。

七、真正遇到“程序卡住”时，先别急着改代码

线上排查最忌讳的是：

程序卡住 ↓ 怀疑死锁 ↓ 凭感觉修改锁 ↓ 重启

这样很容易把最宝贵的现场毁掉。

正确顺序应该是：

程序卡住 ↓ 保存现场 ↓ 抓所有线程栈 ↓ 找到等待位置 ↓ 还原锁依赖 ↓ 找到循环等待 ↓ 再修改代码

其中最重要的一步就是：

抓线程栈。

八、神器一：faulthandler

Python 自带了一个非常实用但经常被忽略的模块：

faulthandler

它可以在超时、信号或者故障发生时输出 Python traceback，而且官方明确说明，其实现能够在 Python 发生死锁时转储调用栈。(Python documentation)

最简单的方式：

importfaulthandler faulthandler.enable()

也可以启动程序时直接：

python-Xfaulthandler app.py

官方还支持通过：

PYTHONFAULTHANDLER

环境变量启用。(Python documentation)

九、让程序卡住 10 秒就自动打印线程栈

排查偶发死锁，我非常推荐：

importfaulthandler faulthandler.dump_traceback_later(timeout=10,repeat=True)

含义是：

10 秒后输出 traceback 如果 repeat=True 以后每隔 10 秒继续输出

于是如果程序真的卡死，你可能看到：

Thread worker-a: File "demo.py", line 16, in worker_a with lock_b: Thread worker-b: File "demo.py", line 27, in worker_b with lock_a: Thread MainThread: File "demo.py", line 38, in <module> t1.join()

这时几乎已经破案：

worker-a → 等 lock_b worker-b → 等 lock_a MainThread → 等 worker-a

如果连续几次 dump 都停在相同位置，死锁嫌疑就非常高。

faulthandler的 traceback 默认输出到sys.stderr，也可以指定日志文件。官方当前实现的转储有帧数和线程数上限，因此超大规模线程程序仍需要配合其他诊断方法。(Python documentation)

十、神器二：sys._current_frames()

如果需要自己编写诊断工具，可以使用：

sys._current_frames()

它会返回：

线程 ID → 当前栈顶 frame

Python 官方甚至直接指出：

这个函数特别适合调试死锁，因为它不需要被死锁线程主动配合。(Python documentation)

可以写一个简单的线程栈检查器：

importsysimportthreadingimporttracebackdefdump_threads():frames=sys._current_frames()forthreadinthreading.enumerate():print(f"\n===== "f"{thread.name}"f"id={thread.ident}"f"native_id={thread.native_id}"f" =====")frame=frames.get(thread.ident)ifframe:traceback.print_stack(frame)

调用：

dump_threads()

就可以看到活跃线程分别卡在哪里。

threading.enumerate()会返回当前活跃的Thread对象，而get_native_id()/Thread.native_id可以帮助把 Python 线程和操作系统线程对应起来。(Python documentation)

这在生产环境中非常实用。

十一、拿到线程栈后怎么看？

不要从第一行开始漫无目的地读。

重点搜索这些位置：

lock.acquire with lock Condition.wait Event.wait Semaphore.acquire queue.get Thread.join Future.result

然后建立一张表：

Thread	已持有	正在等待
worker-a	lock_a	lock_b
worker-b	lock_b	lock_a
main	无	worker-a

接着画等待图：

worker-a │ ▼ lock_b │ ▼ worker-b │ ▼ lock_a │ └────→ worker-a

只要形成闭环，就找到了死锁链条。

这比单纯盯着日志有效得多。

十二、给 acquire() 设置 timeout，避免“永久等待”

默认：

lock.acquire()

可能无限等待。

而 Python 的Lock.acquire()原生支持：

lock.acquire(timeout=...)

如果超过指定时间仍没有获得锁，会返回False。(Python documentation)

例如：

importloggingiflock.acquire(timeout=3):try:process()finally:lock.release()else:logging.error("获取 lock 超时，疑似锁竞争或死锁")

这比：

lock.acquire()

然后永久挂住更容易诊断。

不过要注意：

timeout 不是死锁设计的替代品。

它主要用于：

失败保护 + 故障暴露 + 系统降级

真正的解决方案仍然应该消除循环锁依赖。

十三、不要在持锁状态下做慢 I/O

真实系统中经常出现：

withlock:update_state()requests.get(url)write_database()send_message()

问题是 HTTP、数据库、RPC 都可能非常慢。

如果请求等待 20 秒：

lock 也被占用 20 秒

其他几十个线程可能全部堵住。

这未必是严格意义上的死锁，却会表现得和死锁非常相似：

线程池耗尽 请求越来越慢 最终服务雪崩

更好的设计通常是：

withlock:snapshot=build_snapshot()result=slow_network_call(snapshot)withlock:update_result(result)

即：

锁内只维护共享状态 锁外执行慢操作

当然，如果整个操作必须满足严格事务性，还应该重新设计业务边界，而不是为了缩短锁时间破坏数据一致性。

十四、Condition 用错也可能造成“永久等待”

例如：

condition.wait()

如果负责通知的线程：

condition.notify()

永远没有执行，等待者就可能一直睡下去。

对于带业务条件的场景，更推荐：

withcondition:condition.wait_for(lambda:task_ready,timeout=5)

官方Condition.wait_for()支持 predicate 和 timeout，用来等待某个条件变为真。(Python documentation)

工程上还要同时检查：

谁负责修改 predicate？ 修改时有没有持有正确的 Condition？ 修改后有没有 notify / notify_all？ 等待方是否考虑超时和退出条件？

很多所谓“死锁”，最终其实是条件通知丢失或者状态设计错误。

十五、生产环境建议：给线程和锁“起名字”

线程不要全叫：

Thread-1 Thread-2 Thread-3

而应该：

threading.Thread(target=consume_order,name="order-consumer-01")

Python 当前版本还会在支持的平台上把线程名称设置到操作系统线程名称中，这会让系统级诊断更加方便。(Python documentation)

日志则最好至少包含：

时间 线程名 线程 ID 业务 ID 锁名 操作 耗时

例如：

16:01:03 worker-7 order=83921 acquire inventory_lock 16:01:03 worker-7 acquired inventory_lock 16:01:08 worker-7 waiting payment_lock

出了问题，你就可以直接恢复锁依赖关系。

十六、如何让死锁更容易在测试环境复现？

最难处理的死锁往往是：

生产环境一个月出现一次 本地怎么都复现不了

不要依赖sleep()碰运气。

可以使用：

threading.Barrier

人为控制两个线程同时到达危险位置：

importthreading lock_a=threading.Lock()lock_b=threading.Lock()barrier=threading.Barrier(2)defworker_a():withlock_a:barrier.wait()lock_b.acquire()defworker_b():withlock_b:barrier.wait()lock_a.acquire()

Barrier本身就是让固定数量线程相互等待，直到所有参与线程都抵达屏障以后再继续执行的同步原语。(Python documentation)

这样可以把一个：

“偶尔死锁”

的问题变成：

“稳定复现”

而稳定复现，是解决并发 Bug 的巨大胜利。

十七、工程上如何从源头减少死锁？

我通常推荐下面六条 Python 最佳实践。

第一，统一锁顺序

项目规定：

A → B → C

任何代码都不能：

C → A

第二，尽量减少同时持有多把锁

如果设计可以从：

拿 A 拿 B 修改 释放 B 释放 A

重构成：

锁 A → 获取快照 释放 A 计算 锁 B → 提交结果 释放 B

通常更容易推理。

第三，使用`with`

第四，等待操作尽量考虑 timeout

包括：

lock.acquire(timeout=5)condition.wait(timeout=5)thread.join(timeout=5)

目的不是“用超时治愈死锁”，而是避免系统无声地永久冻结。

第五，减少共享可变状态

如果线程之间本质上是在传递任务：

Producer ↓ Consumer

优先考虑：

queue.Queue

而不是自己组合：

list + Lock + Condition + Event

Python 官方将queue明确定位为线程间安全交换数据的接口。(Python documentation)

第六，把锁当作“架构资源”

锁不是：

哪里报错就在哪里补一个

而应该像数据库事务一样被设计。

一个成熟项目应该能够回答：

这把锁保护什么数据？ 谁可以获取它？ 允许和哪些锁同时持有？ 获取顺序是什么？ 最大持有时间是多少？ 是否允许跨网络 I/O 持锁？ 发生超时时如何降级？

回答不了这些问题，系统规模扩大以后，并发风险通常会迅速增长。

十八、一套可以直接使用的死锁排查流程

当 Python 服务疑似死锁时，我建议按照下面的顺序处理：

① 不要立即重启 ↓ ② 保存当前日志与进程信息 ↓ ③ dump 所有线程 traceback ↓ ④ 找长期停留在 acquire/wait/join 的线程 ↓ ⑤ 写出“已持有锁 → 等待锁” ↓ ⑥ 构造 Wait-For Graph ↓ ⑦ 查找循环依赖 ↓ ⑧ 检查锁获取顺序 ↓ ⑨ 检查嵌套调用是否重复获取 Lock ↓ ⑩ 检查 Condition/Event/Queue 等待关系 ↓ ⑪ 在测试环境构造稳定复现 ↓ ⑫ 重新设计锁顺序或共享状态 ↓ ⑬ 加 timeout、监控和现场转储能力 ↓ ⑭ 写回归测试

千万不要只做：

重启 → 好了 → 工单关闭

因为死锁通常不是服务器“状态不好”。

它是程序同步关系中的结构性问题。

十九、死锁、活锁、饥饿不要混为一谈

最后再区分三个很容易混淆的概念。

死锁 Deadlock

大家都在等待 谁也无法继续

活锁 Livelock

大家都在不断行动 却始终完成不了工作

比如两个线程不断礼让资源：

你先 不，你先 还是你先 ……

CPU 可能还挺忙。

饥饿 Starvation

系统整体一直在运行 但某个线程长期得不到资源

例如高优先级任务不断抢占资源，使某个普通线程始终没有机会执行。

所以：

程序没响应 ≠ 一定死锁

诊断时一定要抓实际线程状态，而不是只凭现象猜测。

二十、总结：排查死锁的核心不是“找卡住的线程”，而是找等待环

Python 死锁看起来复杂，其实可以浓缩为一句话：

某些执行单元占着别人需要的资源，同时又等待别人手中的资源，并最终形成一个无法自行打破的等待环。

因此真正有效的排障思路是：

抓线程栈 ↓ 找到等待点 ↓ 找到资源拥有者 ↓ 继续追踪拥有者正在等待什么 ↓ 构造依赖关系 ↓ 寻找闭环

工具层面，建议至少熟练掌握：

faulthandler sys._current_frames()threading.enumerate()threading.Lock threading.RLock threading.Condition threading.Barrier

其中sys._current_frames()尤其值得记住——Python 官方文档直接把死锁调试列为它最有价值的用途之一。(Python documentation)

而从架构层面，更值得记住四条原则：

统一锁顺序 缩小临界区 减少共享状态 不要无限等待

优秀的 Python 并发程序，不是因为开发者特别擅长“在死锁发生后救火”，而是因为从设计开始，就让等待关系足够简单、透明、可观测。

当有一天你的线上 Python 服务突然安静下来，没有异常、没有报错、没有响应，希望你不会第一时间只想到：

“重启试试。”

而是会想到：

“先把所有线程栈留下来。”

那一刻，你已经从“会使用多线程”，真正迈向了“能够驾驭并发系统”。

延伸阅读

继续深入 Python 并发编程，可以重点阅读 Python 官方threading文档，其中详细介绍了Lock、RLock、Condition、Semaphore、Event 与 Barrier 等同步原语。(Python documentation)

排查程序卡死、崩溃和超时问题，可以重点阅读官方faulthandler文档。(Python documentation)

需要理解线程栈现场抓取时，则建议阅读sys._current_frames()的官方说明。(Python documentation)

**互动思考：**你在实际 Python 项目中遇到过“程序不报错，但就是不往下运行”的情况吗？最后发现的是互斥锁死锁、线程池耗尽、Condition 等待，还是网络 I/O 卡住？欢迎把你的排查思路和典型案例记录下来——并发问题最宝贵的学习材料，往往正来自这些真实的故障现场。