今晚的订单
#04 · 第四晚
危机
因为有团体客人，厨师增加到了六个。可灶台只有三个，而且主厨喊“开业！”之前，谁都不准点火。
目标
开业信号之前没人点火，灶台同时最多只用三个，1 秒内送出六份订单。
深夜食堂 同时
第四晚，九点四十分。因为有团体预订，又来了三位只干一天的厨师。厨师六个，灶台三个。
敏浩第一个站到灶台前点了火。“先热着不是更快嘛。”主厨一言不发地关上了煤气阀。“开业前点火是禁止的。还有，六个人全挤在灶台前，谁也动弹不得。”
秀雅举起手：“在灶台前只挂三张号码牌怎么样？只有拿着号码牌的人才能站到火前面。”主厨头一次笑了。“这就是今晚的第一个工具。”
第 3 课的 Lock 是一支粉笔，也就是一次一个人。Semaphore(n) 是 n 张号码牌。最多 n 个人同时进去，第 n+1 个人要等有人把号码牌还回来。
import threading
import time
stoves = threading.Semaphore(3) # 灶台只有三个
start = time.perf_counter()
def cook(name):
with stoves: # 等到有空灶台再占上
print(f"{time.perf_counter() - start:.1f}s {name} on")
time.sleep(0.3)
names = [f"cook{i}" for i in range(1, 7)]
threads = [threading.Thread(target=cook, args=(n,)) for n in names]
for th in threads:
th.start()
for th in threads:
th.join()
print(f"total {time.perf_counter() - start:.2f}s")
# 例如：
# 0.0s cook1 on
# 0.0s cook2 on
# 0.0s cook3 on
# 0.3s cook4 on
# 0.3s cook5 on
# 0.3s cook6 on
# total 0.60swith stoves: 时拿走一张号码牌（acquire），出来时还回去（release）。号码牌只剩 0 张时，要进去的人就得等。Semaphore(1) 和 Lock 一样，一次只放一个人。不过 Lock 在已释放的状态下再释放会抛 RuntimeError，信号量多调用几次 release() 却会悄无声息地多出号码牌（见下面的 BoundedSemaphore）。在下面的时间线里调一调灶台数。黄色横条是等灶台的时间，橙色横条是真正在做菜的时间。
厨房时间线
订单
6
厨师
3
灶台
2
烹饪
等待
已取消
失败
Semaphore 不会数自己收回了几张。如果因为 bug 多调用了一次 release()，号码牌就悄悄变成了四张。灶台明明只有三个，却有四个人站到了火前面。
loose = threading.Semaphore(3)
loose.release() # 不小心多放了一次
print(sum(loose.acquire(blocking=False) for _ in range(10))) # 灶台变成了四个
# 4
strict = threading.BoundedSemaphore(3)
strict.release()
# ValueError: Semaphore released too many timesBoundedSemaphore 收回的数量一旦超过最初设定的数，就会抛出 ValueError。如果用信号量是为了限制数量，几乎总是 BoundedSemaphore 更好。失误不会悄悄溜过去，而是当场爆出来。
主厨的提醒
信号量只管数人头，可不帮你看着黑板。三个灶台前站了三个人，要是这三个人往同一块黑板上写数字，照样是竞争条件。在最多进三个人的屋子里碰共享值，屋里还得再用一把 Lock。
Event 就是厨房墙上的一盏信号灯。一开始是灭的，有人把它点亮（set()），所有在等这个信号（wait()）的人就会一齐行动。
opened = threading.Event() # 一开始是灭的
start = time.perf_counter()
def cook(name):
opened.wait() # 等信号灯亮
print(f"{time.perf_counter() - start:.1f}s {name} fire")
threads = [threading.Thread(target=cook, args=(n,)) for n in ["minho", "sua", "tak"]]
for th in threads:
th.start()
time.sleep(0.5)
print(opened.is_set())
opened.set() # “开业！”
for th in threads:
th.join()
print(opened.is_set())
# False
# 0.5s minho fire （这三行的顺序每次运行都不一样）
# 0.5s sua fire
# 0.5s tak fire
# True也可以给 wait() 设一个时间上限。这时它的返回值很关键：信号灯亮了就是 True，时间到了还没亮就是 False。
closing = threading.Event()
print(closing.wait(timeout=0.2)) # 等了 0.2 秒，没有信号
closing.set()
print(closing.wait(timeout=0.2)) # 已经亮了就立刻返回 True
closing.clear() # 再关掉
print(closing.is_set())
# False
# True
# False| 方法 | 作用 |
|---|---|
set() | 点亮信号。所有在等的线程全部醒来 |
wait() | 一直等到信号亮起。已经亮着就直接通过 |
wait(timeout) | 最多等这么久。亮了返回 True，否则返回 False |
clear() | 把信号重新关掉。之后的 wait() 又要重新等 |
is_set() | 现在是否亮着 |
有一种用法很常见：用 Event 做一个“营业结束”的信号，让干活的线程每循环一次就用 if closing.is_set(): break 检查一下。这和第 2 课的 daemon 不同，不是中途被掐断，而是让线程自己收拾干净再结束。
下面这条时间线里，主厨会在 0.5 秒后喊“开业！”。
厨房时间线
厨师
3
烹饪
等待
已取消
失败
set() 就会叫醒所有在等的人。Event 只有亮和灭两种状态。可厨房里的等待往往更具体：敏浩在等鸡蛋，秀雅在等豆腐。每当货架上进了食材，就得看一眼“我的到了吗？”，没到就接着等。这种时候用 Condition。
Condition 就是一把锁 + 一张“到了叫我”的名单。它用锁护住共享状态（食材货架），状态一变，就去叫醒名单上的人。
pantry = [] # 食材货架（共享状态）
arrived = threading.Condition() # 护住货架的锁 + “到了叫我”的名单
start = time.perf_counter()
def cook(name, need):
with arrived:
ok = arrived.wait_for(lambda: need in pantry, timeout=0.5)
print(f"{time.perf_counter() - start:.1f}s {name} {need if ok else 'gave up'}")
def supplier():
for item in ["tofu", "egg"]: # 鱼今天不来
time.sleep(0.1)
with arrived:
pantry.append(item)
arrived.notify_all() # 把等着的人全叫醒
threads = [threading.Thread(target=cook, args=a) for a in [("minho", "egg"), ("sua", "tofu"), ("tak", "fish")]]
threads.append(threading.Thread(target=supplier))
for th in threads:
th.start()
for th in threads:
th.join()
# 0.1s sua tofu
# 0.2s minho egg
# 0.5s tak gave upwith arrived: 拿着锁的时候，才去查看和修改货架。这就是第 3 课的规矩。wait_for(条件) 一直等到条件为真。等待期间会暂时放下锁，所以供货商能把食材放上货架。醒来后重新拿到锁，再检查一遍条件。timeout，时间到时就返回条件最后一次的值。鱼始终没来，老卓 0.5 秒后拿到 False，放弃了。notify_all() 会叫醒名单上的所有人。醒来的人各自检查自己的条件，不满足就接着睡。豆腐到的时候，敏浩和老卓也醒了，但不是自己要的食材，于是继续等。notify() 只叫醒一个在等的人。适合谁拿都无所谓的活，比如“做好的一盘菜，随便哪个前厅员工端走都行”。
plates = 0
counter = threading.Condition()
def cook():
global plates
time.sleep(0.1)
with counter:
plates += 1
counter.notify() # 只叫醒一个在等的人
def waiter(name):
global plates
with counter:
counter.wait_for(lambda: plates > 0)
plates -= 1
print(name, "took a plate")
ws = [threading.Thread(target=waiter, args=(n,)) for n in ["w1", "w2"]]
cs = [threading.Thread(target=cook) for _ in range(2)]
for th in ws + cs:
th.start()
for th in ws + cs:
th.join()
print(plates)
# 例如：
# w1 took a plate
# w2 took a plate
# 0一盘菜只叫醒一个人，就不容易出现两个人同时醒来、为一盘菜争抢的情况。反过来，如果每个人等的条件不同，notify() 可能叫醒不相干的人，这时要用 notify_all()。拿不准的话，notify_all() 更保险。
主厨的提醒
在 Condition 里用 wait() 的时候，一定要放在 while 里重新检查条件。被叫醒不代表你的食材到了。可能是别人的，也可能被人先拿走了。用 wait_for(条件)，它会替你转那个 while，最省事。
团体 4 人桌的四盘菜必须同时上。要是只先上一盘，剩下三个人就得干坐着，看着菜变凉。Barrier(n) 会在出菜口等到 n 个人全部到齐，凑齐了再一起放行。
start = time.perf_counter()
table = threading.Barrier(4, action=lambda: print("serve 4 plates"))
done = []
def cook(name, sec):
time.sleep(sec) # 每个人速度不一样
i = table.wait() # 在出菜口等四个人到齐
done.append((round(time.perf_counter() - start, 1), i, name))
crew = [("minho", 0.1), ("sua", 0.2), ("chef", 0.3), ("tak", 0.4)]
threads = [threading.Thread(target=cook, args=c) for c in crew]
for th in threads:
th.start()
for th in threads:
th.join()
print(sorted(done))
# serve 4 plates
# [(0.4, 0, 'minho'), (0.4, 1, 'sua'), (0.4, 2, 'chef'), (0.4, 3, 'tak')]wait() 会给每个人返回一个 0 到 n-1 之间互不相同的编号。在 CPython 里就是按到达顺序排的 0、1、2、3。可以用这个编号来指定“只需一个人做的事”（比如 if i == 0: 时记日志）。action= 的函数，只在最后一个人到达时执行一次。这就是上菜的信号。wait(timeout=) 超时的话，Barrier 就会被打破，所有在等的人都会收到 BrokenBarrierError。| 工具 | 换成厨房来说 | 什么时候用 |
|---|---|---|
Lock | 一支粉笔 | 共享值一次只让一个人改（第 3 课） |
Semaphore(n)、BoundedSemaphore(n) | 灶台前的 n 张号码牌 | 最多同时用 n 个 |
Event | 墙上的信号灯 | 把“开始”“结束”这类一次性信号发给所有人 |
Condition | 锁 + “到了叫我”名单 | “满足这个条件时叫醒我”。等共享状态发生变化时 |
Barrier(n) | 团体桌的出菜口 | n 个人到齐后一起出发 |
这五样都在 threading 里，也都是在第 3 课的锁之上搭起来的工具。实际上，Semaphore、Event、Barrier 内部用的是 Condition，而 Condition 内部握着一把锁（不另外指定的话就是 RLock）。
今晚的目标是：开业信号之前没人点火，灶台同时最多只用三个，1 秒内送出六份订单。用 Event 定起跑线，用 Semaphore(3) 管灶台，用 Lock 数“现在火前站着几个人”。
opened = threading.Event()
stoves = threading.Semaphore(3)
lock = threading.Lock()
on_fire = peak = 0
log = []
start = time.perf_counter()
def cook(n):
global on_fire, peak
opened.wait() # 1. “开业！”之前不点火
with stoves: # 2. 灶台同时最多三个
with lock:
on_fire += 1
peak = max(peak, on_fire)
time.sleep(0.3)
with lock:
on_fire -= 1
log.append((round(time.perf_counter() - start, 1), f"#{n}"))
threads = [threading.Thread(target=cook, args=(n,)) for n in range(1, 7)]
for th in threads:
th.start()
time.sleep(0.3) # 准备迎客
opened.set()
for th in threads:
th.join()
print(sorted(log))
print(f"peak {peak}, total {time.perf_counter() - start:.2f}s")
# 例如：
# [(0.6, '#1'), (0.6, '#2'), (0.6, '#3'), (0.9, '#4'), (0.9, '#5'), (0.9, '#6')]
# peak 3, total 0.90speak 跑多少次都是 3。0.6 秒出来的是哪三个订单，每次运行可能不一样。on_fire 和 peak 是六个厨师一起改的值，所以用 lock 包了起来。哪怕已经在信号量里面，也要这么做。快打烊时，一位独自坐在靠窗位子的女客人吃完了一碗面。她看了一眼手表，在小本子上记了点什么，然后安静地结账离开了。
主厨一个个关着灶台的阀门说：“今天大家排队排得不错。可要是除了灶台，刀和砧板也各只有一份呢？敏浩拿着刀等砧板，秀雅拿着砧板等刀，会怎么样？”
营业日志 — 第四晚。厨师 6 人，灶台 3 个。开业信号（Event）之前待命，灶台用三张号码牌（Semaphore）限制，0.9 秒完成 6 单。靠窗的客人一位，面一碗。
closing = threading.Event()，让厨师线程每 0.1 秒做一碗，一旦 closing.is_set() 就停下。主线程在 0.35 秒后调用 closing.set()，然后 join()。一共做出了几碗？Barrier(4) 改成 Barrier(5) 会发生什么。为了不让它真的永远卡住，改成 table.wait(timeout=1) 再运行，看看会抛出什么异常。主厨的配方卡
Semaphore(n)：用 with sem: 限制最多同时 n 个。用于限制数量时，换成 BoundedSemaphore(n)，release 失误会以 ValueError 暴露出来。
Event：set() 点亮后，所有在 wait() 的线程都会醒来。wait(timeout) 亮了返回 True，否则 False。还有 clear()、is_set()。
Condition：在 with cond: 里调用 cond.wait_for(条件)。改了状态的一方调用 notify()（一个）或 notify_all()（全部）。
Barrier(n)：wait() 等 n 个人到齐一起通过。返回编号 0~n-1，action= 只执行一次。
就算在信号量里面，改共享值也要用 Lock。工具是叠起来用的。
订阅新课时和新课程通知
本课程新增课时或推出后续课程时，我们会发邮件通知你。
问答
关于本课有任何疑问都可以在这里留言。所有语言版本页面的提问会汇集在一起。