锁顺序问题是python死锁最常见原因,指多个线程以不一致顺序获取同一组锁(如线程1先a后b、线程2先b后a),导致互相等待永久阻塞。

Python中死锁最常见原因之一是多个线程以不同顺序获取多个锁,导致互相等待——即“锁顺序问题”。只要两个或以上线程对同一组锁的加锁顺序不一致,就可能陷入永久阻塞。
什么是锁顺序问题
假设有两个锁 lock_A 和 lock_B。线程1先获取 lock_A,再尝试获取 lock_B;而线程2恰好相反:先获取 lock_B,再尝试获取 lock_A。当线程1拿到 lock_A、线程2拿到 lock_B 后,两者都会卡在第二个锁上,谁也无法继续,形成死锁。
典型触发场景
- 多个函数各自独立加锁,但未约定全局加锁顺序(例如:函数A按 A→B 加锁,函数B按 B→A 加锁)
- 嵌套调用时,底层函数悄悄加了新锁,调用方 unaware,无意中改变了锁序
- 使用可重入锁(RLock)误以为安全,但跨锁协作时仍需严格顺序
- 分布式或模拟多资源场景(如转账:账户A扣款→账户B入账),资源顺序不统一
如何避免锁顺序问题
- 固定全局锁顺序:对所有涉及的锁定义唯一编号或排序规则(如按锁对象 id()、变量名字符串排序),始终按升序(或降序)获取
- 一次性申请所有锁:用 threading.Lock 的 acquire(timeout=...) 配合循环重试,或使用 contextlib.ExitStack 管理多锁,但更推荐用 concurrent.futures.ThreadPoolExecutor 减少手动锁需求
- 避免嵌套加锁:设计函数时明确是否持有锁;若必须嵌套,确保被调函数不引入新锁,或由最外层统一管理
- 用超时机制兜底:给 lock.acquire(timeout=0.5) 设置合理超时,失败后释放已持锁并重试/报错,防止无限等待
一个可复现的死锁示例
以下代码会大概率触发死锁:
import threading
import time
<p>lock_a = threading.Lock()
lock_b = threading.Lock()</p><p>def thread1():
with lock_a:
time.sleep(0.1)
with lock_b: # 等待 lock_b
print("Thread1 done")</p><p>def thread2():
with lock_b:
time.sleep(0.1)
with lock_a: # 等待 lock_a
print("Thread2 done")</p><p>t1 = threading.Thread(target=thread1)
t2 = threading.Thread(target=thread2)
t1.start(); t2.start()
t1.join(); t2.join()</p>运行后程序常卡住不动——因为 thread1 持有 lock_a 等 lock_b,thread2 持有 lock_b 等 lock_a,双方僵持。
立即学习“Python免费学习笔记(深入)”;
修复思路很简单
强制统一顺序:比如规定所有地方都先 lock_a 再 lock_b。修改任一函数即可打破循环依赖:
def thread2_fixed():
with lock_a: # 先 a
with lock_b: # 再 b
print("Thread2 fixed")
或者更健壮地用排序逻辑自动保证顺序,适合锁数量多、来源动态的场景。










