今晚的订单
#05 · 第五晚
危机
敏浩攥着刀等砧板，秀雅抱着砧板等刀。谁都不打算先松手。厨房停住了。
目标
三个同时要用刀和砧板的厨师，各做 100 次菜，一次都不卡住。
深夜食堂 同时
第五晚，十一点。厨房安静得出奇。敏浩握着刀，眼睛盯着砧板；秀雅抱着砧板，眼睛盯着刀。两个人翻来覆去只有一句：“我用完就给你。”
老卓切葱切到一半，嘟囔了一句：“两个人都站着不动，谁也做不完。”
尹瑞妍主厨按停了脖子上挂的计时器。“都放下。这不是谁手慢的问题，是没有规矩的问题。”
和昨晚的灶台规矩一样，刀和砧板也是一次只能一个人用的锁。敏浩按 刀 → 砧板 的顺序拿，秀雅按 砧板 → 刀 的顺序拿。
import threading
import time
knife = threading.Lock()
board = threading.Lock()
def minho():
with knife: # 拿起刀
time.sleep(0.1)
with board: # 等砧板
print("minho done")
def sua():
with board: # 拿起砧板
time.sleep(0.1)
with knife: # 等刀
print("sua done")
a = threading.Thread(target=minho, daemon=True)
b = threading.Thread(target=sua, daemon=True)
a.start(); b.start()
a.join(timeout=1); b.join(timeout=1)
print("alive:", a.is_alive(), b.is_alive())
# alive: True Truejoin(timeout=1) 最多等 1 秒就返回。线程设成了 daemon=True，所以程序照常结束。自己动手做死锁实验时，一定要像这样给自己留一条退路。time.sleep(0.1) 是为了让两个人几乎同时拿起第一件工具。真实程序里这个空档极少出现，于是死锁变成“偶尔才发作”的 bug，更难缠。想在程序里察觉“卡住了”，就用 acquire(timeout=...)。拿到了返回 True，限定时间内没拿到返回 False。with lock: 没法设时间限制，这时就直接调用 acquire 和 release。
knife = threading.Lock() # 换一把新刀、一块新砧板
board = threading.Lock()
def cook(name, first, second, patience):
with first:
time.sleep(0.1)
if second.acquire(timeout=patience): # 最多等 patience 秒
second.release()
print(name, "done")
else:
print(name, f"gave up after {patience}s")
a = threading.Thread(target=cook, args=("minho", knife, board, 1.0))
b = threading.Thread(target=cook, args=("sua", board, knife, 0.5))
a.start(); b.start()
a.join(); b.join()
# sua gave up after 0.5s
# minho done秀雅等了 0.5 秒就放弃，放下了砧板；还在等的敏浩拿到砧板，把菜做完了。只要有一个人让一步，死锁就解开了。为什么？下一节的四个条件会给出答案。
只有下面四个条件同时全部满足，才会发生死锁。反过来说，只要打破其中一个，死锁就不会出现。
| 条件 | 含义 | 今晚的厨房里 | 怎么打破 |
|---|---|---|---|
| 互斥 | 资源一次只能一个人用 | 刀只能在一个人手里 | 多买几把刀，或者设计成不需要共享 |
| 占有并等待 | 攥着一样，等另一样 | 拿着刀等砧板 | 两样一次拿齐，或者拿不到就放下 |
| 不可剥夺 | 别人手里的资源不能强行拿走 | 不能从敏浩手里抢刀 | 用时间限制让人自己放下（解法 2） |
| 循环等待 | 等待关系连成一个圈 | 敏浩 → 秀雅 → 敏浩 | 所有人按同一顺序拿（解法 1） |
主厨的提醒
死锁不吃 CPU，不报错，也不留日志。就那么悄悄地停着。所以一有人报“变慢了”，我头一个怀疑的就是死锁。
主厨在黑板上写下规矩：“每件工具编个号。刀 1 号，砧板 2 号。永远先拿编号小的。”
敏浩和秀雅都先拿刀的话，只有拿到刀的人才会去等砧板。没拿到刀的人两手空空地等着，圈就连不起来。别把规矩交给人的记性，把它写进一个函数里。
from contextlib import contextmanager
KNIFE = (1, threading.Lock()) # (编号, 锁)
BOARD = (2, threading.Lock())
@contextmanager
def hold(*tools):
ordered = sorted(tools, key=lambda t: t[0]) # 编号小的先拿
for _, lock in ordered:
lock.acquire()
try:
yield
finally:
for _, lock in reversed(ordered): # 放的时候倒过来
lock.release()
done = []
def minho():
for _ in range(100):
with hold(KNIFE, BOARD):
time.sleep(0)
done.append("minho")
def sua():
for _ in range(100):
with hold(BOARD, KNIFE): # 写的顺序不同，拿的顺序一样
time.sleep(0)
done.append("sua")
ts = [threading.Thread(target=f, daemon=True) for f in (minho, sua)]
for t in ts: t.start()
for t in ts: t.join(timeout=2)
print(sorted(done), [t.is_alive() for t in ts])
# ['minho', 'sua'] [False, False]hold(BOARD, KNIFE)，hold 里面也会按编号排序，实际还是 刀 → 砧板 的顺序。time.sleep(0) 故意留出交接的空档，也一样。id(lock) 这类不会重复的值。关键是所有代码用同一套标准。一旦有地方需要两把以上的锁，就把这条规矩写在代码最上方的注释或文档里。新来的人不知道顺序、反着拿的那一刻，死锁就又回来了。
有时候顺序很难定，比如事先不知道会用到哪些锁。这时就用另一条规矩：“等了一段时间还拿不到，就连手里的一起放下，从头再来。”
import random
tries = {}
def polite_cook(name, first, second):
tries[name] = 0
while True:
tries[name] += 1
with first:
time.sleep(0.05)
if second.acquire(timeout=0.1): # 0.1 秒内拿不到就算了
try:
return # 两样都拿到了：菜做完
finally:
second.release()
# 离开 with 时已经放下了 first
time.sleep(random.uniform(0, 0.1)) # 随机歇一会儿再试
a = threading.Thread(target=polite_cook, args=("minho", knife, board))
b = threading.Thread(target=polite_cook, args=("sua", board, knife))
a.start(); b.start()
a.join(); b.join()
print(tries)
# 例：{'minho': 2, 'sua': 3}（每次运行都不一样）撞上几次之后，两个人最终都能做完。可要是把 random.uniform(0, 0.1) 去掉，就可能冒出一个新问题：活锁（livelock）。
| 时刻 | 敏浩 | 秀雅 |
|---|---|---|
| 1 | 拿起刀 | 拿起砧板 |
| 2 | 没有砧板 → 放下刀 | 没有刀 → 放下砧板 |
| 3 | 同样歇 0.1 秒，再拿起刀 | 同样歇 0.1 秒，再拿起砧板 |
| … | 1~3 无限循环 | 1~3 无限循环 |
就像走廊里迎面碰上的两个人，总往同一边让。和死锁不同，两个人都忙个不停，却谁也做不完菜。把歇的时间打乱成随机的，总有一个人会先动，循环就被打破了。
真实线程的执行时机多少会有抖动，所以用简短的例子很难每次都复现活锁。正因如此才更危险：它只在负载高的日子冒出来。用解法 2 时，一定要配上随机等待。
计算机科学里有个著名的哲学家就餐问题。圆桌旁坐着五个人，每两人之间放一把叉子。要吃饭，左右两边的叉子都得拿到。搬进厨房，就是圆形料理台前的五个厨师和五把刀。
N = 5
def service(ordered):
knives = [threading.Lock() for _ in range(N)] # 五把刀
meals = [0] * N
def cook(i):
left, right = i, (i + 1) % N
if ordered:
first, second = min(left, right), max(left, right)
else:
first, second = left, right
for _ in range(3):
with knives[first]:
time.sleep(0.05) # 拿着第一把刀停一下
with knives[second]:
meals[i] += 1
ts = [threading.Thread(target=cook, args=(i,), daemon=True) for i in range(N)]
for t in ts: t.start()
for t in ts: t.join(timeout=1)
return sum(t.is_alive() for t in ts), meals
print("left first:", service(ordered=False))
print("small first:", service(ordered=True))
# left first: (5, [0, 0, 0, 0, 0])
# small first: (0, [3, 3, 3, 3, 3])下面的实验台会一步一步地演示这个场景。要是真让服务器上的线程陷入死锁，就再也解不开了，所以唯独这个实验台是模拟运行的。刀上的数字是握着那把刀的厨师编号。
刀与砧板
1
2
3
4
5
·
·
·
·
·
步数 0
完成的菜: 1:0 2:0 3:0 4:0 5:0
程序卡住了，却不知道卡在哪里？标准库的 faulthandler 能帮忙。它会打印出每个线程此刻停在第几行。把下面的代码存成 stuck.py 运行看看。
import faulthandler
import threading
import time
knife = threading.Lock()
board = threading.Lock()
def minho():
with knife:
time.sleep(0.1)
with board:
pass
def sua():
with board:
time.sleep(0.1)
with knife:
pass
faulthandler.dump_traceback_later(1) # 1 秒后还在这里，就打印所有线程
crew = [threading.Thread(target=f, name=f.__name__, daemon=True) for f in (minho, sua)]
for t in crew: t.start()
for t in crew: t.join(timeout=2)
faulthandler.cancel_dump_traceback_later() # 按时结束的话就取消预约Timeout (0:00:01)!
Thread 0x00009560 [sua] (most recent call first):
File "stuck.py", line 17 in sua
...
Thread 0x0000a670 [minho] (most recent call first):
File "stuck.py", line 11 in minho
...输出做了删减（地址和路径每次运行都不一样）。秀雅停在第 17 行 with knife:，敏浩停在第 11 行 with board:。谁在等什么，一目了然。
faulthandler.dump_traceback_later(timeout)：预约在 timeout 秒后，把所有线程的位置打印到标准错误。repeat=True 会一直重复，exit=True 会在打印后结束程序。faulthandler.dump_traceback(all_threads=True)：不用预约，马上打印。适合在耗时很长的任务中途看一眼状态。dump_traceback_later。卡住的那天，“凶手”就留下了。主厨的提醒
拿着锁去调别的函数时要小心。你又不知道那个函数里面还会拿什么锁。锁里面只干像往黑板上写个数那样又短又明白的事。这条比顺序规矩还要紧。
今晚的目标是：三个同时要用刀和砧板的厨师，各做 100 次菜，一次都不卡住。老卓连锅一起要用三样。给每件工具编号、统一用 hold 来拿，那么谁按什么顺序写都无所谓。
PAN = (3, threading.Lock())
meals = {}
def cook(name, *tools):
for _ in range(100):
with hold(*tools): # 永远先拿编号小的
time.sleep(0)
meals[name] = meals.get(name, 0) + 1
crew = [threading.Thread(target=cook, args=a, daemon=True) for a in [
("minho", KNIFE, BOARD), ("sua", BOARD, KNIFE), ("tak", PAN, BOARD, KNIFE)]]
for t in crew: t.start()
for t in crew: t.join(timeout=3)
print([meals[n] for n in ("minho", "sua", "tak")], sum(t.is_alive() for t in crew))
# [100, 100, 100] 0三个人各做了 100 次，卡住的人数是 0。主厨在黑板的规矩旁边又添了一行：“刀 1，砧板 2，锅 3。编号小的先拿。”
“今天是因为工具卡住了。明天会因为人卡住。订单雪片一样飞进来的时候，谁做哪一单，由谁来定？”
营业日志 — 第五晚。刀和砧板被人按相反顺序拿，厨房卡死。给工具编号、约定先拿编号小的之后，三个人各做 100 次，一次没卡。
time.sleep(0.1) 删掉，多运行几次。每次都会死锁吗？用一句话写下：为什么“偶尔才卡住的 bug”更危险。service 里用 print 确认：ordered=True 时，只有 4 号厨师换了拿刀的顺序。再用四个条件解释：为什么只改一个厨师的顺序，死锁就解开了。主厨的配方卡
死锁的四个条件：互斥、占有并等待、不可剥夺、循环等待。打破一个就能解开。
解法 1：给锁编号，所有人都先拿编号小的。用 sorted(tools) 写进一个函数里。
解法 2：if not lock.acquire(timeout=0.1): 放下手里的东西，随机歇一会儿再试。
活锁：所有人踩着同一个节拍只顾着让，谁也做不完。用随机等待打破。
做实验时用 daemon=True + join(timeout=1) + is_alive() 给自己留退路。
找卡住的位置：faulthandler.dump_traceback_later(1)、dump_traceback(all_threads=True)。
订阅新课时和新课程通知
本课程新增课时或推出后续课程时，我们会发邮件通知你。
问答
关于本课有任何疑问都可以在这里留言。所有语言版本页面的提问会汇集在一起。