本文目录导读:

- 破除“循环等待”:固定锁的获取顺序
- 破除“持有并等待”:一次性获取所有锁(TryLock 模式)
- 使用“超时机制”(Timeout)
- 使用可重入锁(RLock/RentrantLock)
- 使用更高级的抽象
- 锁排序与分层(适用于复杂系统)
- 针对不同脚本语言的注意点:
- 一条简单的黄金法则
在脚本中避免死锁,核心在于破坏死锁的四个必要条件(互斥、持有并等待、不可剥夺、循环等待),由于脚本语言(如 Python、JavaScript、Shell)通常运行在单线程或拥有 GIL(全局解释器锁)的环境中,多线程死锁常见于使用 Lock、RLock、Semaphore 等同步原语时。
以下是避免死锁的几种经典策略,附带脚本语言(以 Python 为例)的代码示例和注意点:
破除“循环等待”:固定锁的获取顺序
这是最常用且最可靠的方法,如果多个线程需要获取多个锁,强行规定所有线程必须以相同的顺序获取锁,这样就不会形成“A等B的锁,B等A的锁”的环。
不做(会死锁):
import threading
lock_a = threading.Lock()
lock_b = threading.Lock()
def task_1():
with lock_a:
# 模拟操作
with lock_b: pass
def task_2():
with lock_b: # 顺序与 task_1 相反!
with lock_a: pass
做(避免死锁):
# 强制所有线程先获取 lock_a,再获取 lock_b
def task_1():
with lock_a:
with lock_b: pass
def task_2():
with lock_a: # 与 task_1 顺序一致
with lock_b: pass
破除“持有并等待”:一次性获取所有锁(TryLock 模式)
允许线程尝试获取锁,如果无法立即获取所有需要的锁,则释放已经持有的锁,稍后重试,这避免了线程占据着一个锁去等待另一个锁。
Python 示例(使用 acquire(timeout) 或 try: ... finally:):
import threading
import time
lock_a = threading.Lock()
lock_b = threading.Lock()
def worker():
while True:
# 尝试获取锁 A
if lock_a.acquire(timeout=0.1):
print("Got lock A")
# 尝试获取锁 B
if lock_b.acquire(timeout=0.1):
print("Got lock B, doing work")
# 成功获取所有锁,执行任务
time.sleep(0.5)
lock_b.release()
lock_a.release()
break
else:
# 无法获得 B,释放 A,退避重试
print("Can't get B, release A")
lock_a.release()
time.sleep(0.1) # 退避
else:
time.sleep(0.1) # 退避
使用“超时机制”(Timeout)
在获取锁时设置超时时间,如果超过该时间未能获取锁,就放弃此次操作,并释放已经持有的锁(如果有),避免线程无限期阻塞。
Python acquire(timeout) 是首选:
if lock.acquire(timeout=5): # 最多等待5秒
try:
# 执行临界区
pass
finally:
lock.release()
else:
print("获取锁超时,执行其他逻辑或重试")
使用可重入锁(RLock/RentrantLock)
如果同一个线程可能会递归地获取同一把锁(例如函数 A 调用函数 B,两者都需要同一把锁),使用 RLock 而不是 Lock。Lock 会导致死锁,而 RLock 允许同一线程多次获取(通过引用计数),解决了这种情况下的“自我死锁”。
lock = threading.RLock()
def a():
with lock:
b() # 递归调用,如果是 Lock 会死锁
def b():
with lock: # RLock 允许,因为当前线程已持有
pass
使用更高级的抽象
尽量不使用底层的锁,而是使用无需手动加锁的高级组件,它们内部已经处理了死锁问题:
- 队列(Queue):让线程通过
queue.put()/queue.get()通信,而不是共享状态加锁。 - 线程池(ThreadPoolExecutor):聚焦于任务提交,避免手动锁管理。
- Actor 模型 / 异步(Async/Await):用消息传递代替共享内存,从根本上避免传统锁死锁。
锁排序与分层(适用于复杂系统)
如果涉及多种锁(如资源锁、日志锁、数据库连接锁),给锁分配唯一等级,规则是:线程只能获取等级比当前持有的锁更高或更低的锁,不能交错。
针对不同脚本语言的注意点:
| 语言/环境 | 核心建议 |
|---|---|
| Python | GIL 下多线程死锁主要发生在 Lock/RLock,优先考虑固定顺序和 timeout。 |
| JavaScript | 单线程、事件循环,不存在传统死锁,但 async/await 中可能出现“死锁”假象:await 了一个永远不会 resolve 的 Promise,或两个协程互相 await,解决方法:避免循环 Promise 依赖,使用 Promise.race 加超时。 |
| Shell | 死锁常见于管道 + 文件锁(flock)。 |
| Lua | 协程死锁模式与 JS 类似,C API 加锁时注意与 GC hook 的交互。 |
一条简单的黄金法则
“所有线程必须以相同的顺序获取所有锁,否则必须使用退避 / 超时模式。”
如果代码中出现了多个 with lock_a: with lock_b: 和 with lock_b: with lock_a: 同时存在,几乎一定会死锁,建议在团队内制定锁获取顺序的规范,并在代码审查时重点检查。