今晚的订单
#06 · 第六晚
危机
前厅的订单一张接一张地涌进来，敏浩和秀雅同时抓起同一张订单，老卓却闲着没事干。谁做哪一单，没人定过。
目标
12 份订单由三个厨师分着做，每份不多不少只做一次。并且确切地知道全部做完了。
深夜食堂 同时
第六晚，星期五。前厅员工拿着订单冲进厨房，见人就塞。敏浩和秀雅同时抓起同一张订单，结果同一碗面上了两份。这期间，老卓握着刀，干等着活儿上门。
尹瑞妍主厨在厨房门口上方钉了一根长铁杆，这就是订单夹。“前厅只管往这儿夹。厨师只管从这儿取。互相别搭话。”
把产生订单的一方（前厅员工）和处理订单的一方（厨师）分开，中间放一条队列，这种设计叫生产者-消费者模式。在 Python 里，这条队列就是 queue.Queue。
import queue
rail = queue.Queue() # 订单夹
for n in range(1, 5):
rail.put(f"#{n}") # 夹上去
print(rail.qsize(), rail.get(), rail.get(), rail.qsize())
# 4 #1 #2 2先夹上去的订单先出来（FIFO，先进先出）。现在用线程跑一个前厅员工和两个厨师。
import threading
import time
rail = queue.Queue()
served = []
def hall(): # 生产者：接单，夹上去
for n in range(1, 7):
time.sleep(0.05)
rail.put(f"#{n}")
def cook(name): # 消费者：从订单夹取下来做
for _ in range(3):
order = rail.get() # 订单夹空着，就在这里等
time.sleep(0.1)
served.append((name, order))
ts = [threading.Thread(target=hall),
threading.Thread(target=cook, args=("minho",)),
threading.Thread(target=cook, args=("sua",))]
for t in ts: t.start()
for t in ts: t.join()
print(sorted(order for _, order in served))
# ['#1', '#2', '#3', '#4', '#5', '#6']Lock 都没有。Queue 内部已经装好了锁和第 4 课的 Condition，put 和 get 会自动保证一次只处理一个人。队列替你上了锁。get() 会一直等到有订单夹上来。厨师不必一遍遍地问“有单吗？”，原地空转。线程之间需要传递数据时，先想想能不能用队列递过去，而不是给共享变量加锁。直接摆弄锁的代码越少，第 3 课和第 5 课那样的事故就越少。
上面的例子给每个厨师定好了“做三份就下班”。可真正的厨房里，没人知道今晚会来多少单。厨师一直盯着订单夹，主厨只想知道一件事：夹上去的订单是不是全做完了。
rail = queue.Queue()
def cook(name):
while True:
order = rail.get()
time.sleep(0.1)
rail.task_done() # “这张订单做完了”
for name in ("minho", "sua", "tak"):
threading.Thread(target=cook, args=(name,), daemon=True).start()
start = time.perf_counter()
for n in range(1, 10):
rail.put(f"#{n}")
rail.join() # 等到夹上去的订单全部做完
print(f"all 9 done: {time.perf_counter() - start:.2f}s")
# all 9 done: 0.30s队列内部记着一个数：还没做完的订单数。
| 操作 | 未完成的订单数 |
|---|---|
调用一次 put() | 加 1 |
调用一次 task_done() | 减 1 |
join() | 一直等到它变成 0 |
get() 取出来不等于做完了。菜做好之后调用 task_done()，这个数才会减少。while True 永远盯着订单夹，所以设成了 daemon=True。订单全部做完后 join() 放行，程序结束时厨师们也跟着下班。task_done() 调用的次数比取出的订单还多，就会报 ValueError: task_done() called too many times。主厨的提醒
漏了 task_done，join 就永远放不了行。做菜时抛了异常也一样。所以我总是用 try/finally 包起来：'finally: rail.task_done()'。盘子摔了，订单也得放下。
订单夹的格数是可以定的。Queue(maxsize=2) 就是最多只能夹两张订单。往满了的订单夹上 put()，放的一方就得等。相当于厨房一忙不过来，前厅就放慢接单的速度。
rail = queue.Queue(maxsize=2) # 最多只能夹两张订单
rail.put("#1")
rail.put("#2")
print(rail.full())
# True
try:
rail.put("#3", timeout=0.5) # 等了 0.5 秒还没空位的话
except queue.Full:
print("Full after 0.5s")
# Full after 0.5s
try:
rail.put_nowait("#3") # = put("#3", block=False)
except queue.Full:
print("Full right away")
# Full right away
print(rail.get_nowait(), rail.get(timeout=0.5))
# #1 #2
try:
rail.get(timeout=0.5) # 订单夹空了
except queue.Empty:
print("Empty after 0.5s")
# Empty after 0.5s| 调用方式 | 订单夹满了／空了的时候 |
|---|---|
put(x)、get() | 一直等到有空位（有订单来） |
put(x, timeout=0.5)、get(timeout=0.5) | 最多等 0.5 秒，然后抛 queue.Full / queue.Empty |
put_nowait(x)、get_nowait() | 不等，立刻抛 queue.Full / queue.Empty |
不给 maxsize 或者给 0，订单夹就可以无限长。如果前厅比厨房快得多，订单就会无休止地堆积，把内存塞满。定好容量，快的一方就会配合慢的一方等着。这叫背压（backpressure）。
rail = queue.Queue(maxsize=2)
def cook(): # 一个慢吞吞的厨师
while True:
rail.get()
time.sleep(0.1)
rail.task_done()
threading.Thread(target=cook, daemon=True).start()
start = time.perf_counter()
for n in range(1, 7):
rail.put(f"#{n}") # 前厅手快：没空位就在 put 这儿等
print(f"hall done at {time.perf_counter() - start:.1f}s")
rail.join()
print(f"kitchen done at {time.perf_counter() - start:.1f}s")
# hall done at 0.3s
# kitchen done at 0.6s前厅本想一眨眼就把六张订单全夹上去，可订单夹只有两格，只好在 put 上等了 0.3 秒。去掉 maxsize 的话，前厅马上就完事，订单则在夹子上越堆越多。
在下面的时间线上亲自看看吧。上面几行是前厅员工，下面几行是厨师。前厅员工接一单要 0.1 秒（橙色），等订单夹腾出空位的时间是黄色。
厨房时间线
订单
6
前厅员工
1
厨师
3
单夹容量
3
烹饪
等待
已取消
失败
与其用 daemon=True 把厨师撂在那儿不管，有时我们更想通知一声“今天打烊了”，让他们体面地下班。传统的做法是哨兵（sentinel）：夹上一张写着“打烊”的特殊订单。
STOP = None # 哨兵：“打烊”订单
rail = queue.Queue()
served = []
def cook(name):
while True:
order = rail.get()
if order is STOP:
break # 拿到哨兵就下班
time.sleep(0.05)
served.append(order)
cooks = [threading.Thread(target=cook, args=(n,)) for n in ("minho", "sua")]
for t in cooks: t.start()
for n in range(1, 7):
rail.put(f"#{n}")
for _ in cooks:
rail.put(STOP) # 有几个厨师就夹几张
for t in cooks: t.join()
print(len(served), [t.is_alive() for t in cooks])
# 6 [False, False]哨兵必须按厨师的人数来夹。只夹一张，就只有一个人下班，其余的人会在空订单夹前永远等下去。订单夹是 FIFO，所以哨兵要等前面夹上的订单全部取完才会出来。
从 Python 3.13 开始，可以直接把订单夹关掉。调用 Queue.shutdown() 之后，再 put() 会抛 queue.ShutDown；get() 则会先把剩下的订单都交出去，然后再抛 queue.ShutDown。不用再数厨师有几个了。
rail = queue.Queue()
served = []
def cook(name):
while True:
try:
order = rail.get()
except queue.ShutDown: # 订单夹关了，也没有剩下的订单
return
time.sleep(0.05)
served.append(order)
rail.task_done()
cooks = [threading.Thread(target=cook, args=(n,)) for n in ("minho", "sua")]
for t in cooks: t.start()
for n in range(1, 7):
rail.put(f"#{n}")
rail.shutdown() # 不再接新单。已夹上的订单照样做完
for t in cooks: t.join()
print(len(served))
# 6
try:
rail.put("#7")
except queue.ShutDown:
print("rail is closed")
# rail is closed急着关门的时候，用 shutdown(immediate=True)。它会把订单夹上剩下的订单全部扔掉，正在等待的 get() 也会立刻抛出 queue.ShutDown。
rail = queue.Queue()
served = []
def cook(name):
while True:
try:
order = rail.get()
except queue.ShutDown:
return
time.sleep(0.1)
served.append(order)
rail.task_done()
cooks = [threading.Thread(target=cook, args=(n,)) for n in ("minho", "sua")]
for t in cooks: t.start()
for n in range(1, 7):
rail.put(f"#{n}")
time.sleep(0.15) # 两个人正在做 #3 和 #4
rail.shutdown(immediate=True) # 扔掉夹上剩下的 #5、#6，马上关闭
rail.join() # 手上的 #3、#4 做完就放行
for t in cooks: t.join()
print(sorted(served), rail.qsize())
# ['#1', '#2', '#3', '#4'] 0shutdown() | shutdown(immediate=True) | |
|---|---|---|
新的 put() | queue.ShutDown | queue.ShutDown |
| 已经夹上的订单 | 照样交出去 | 扔掉（也从未完成订单数里扣掉） |
get() | 订单夹空了之后抛 queue.ShutDown | 立刻抛 queue.ShutDown |
| 厨师手上的订单 | 做完为止 | 做完为止 |
immediate=True 也不会把厨师已经拿在手里的订单抢走。所以 join() 要等这些订单的 task_done() 都调用过之后才放行。
queue 模块里还有几种取出顺序不同的订单夹。
PriorityQueue：从小的值开始取。想先做 VIP 桌的订单时就用它。惯例是放入 (优先级, 编号, 订单) 这样的元组。
import itertools
vip_first = queue.PriorityQueue()
ticket = itertools.count(1) # 优先级相同就按先来后到
for prio, dish in [(2, "noodle"), (1, "VIP soup"), (2, "rice"), (1, "VIP noodle")]:
vip_first.put((prio, next(ticket), dish))
while not vip_first.empty():
print(vip_first.get())
# (1, 2, 'VIP soup')
# (1, 4, 'VIP noodle')
# (2, 1, 'noodle')
# (2, 3, 'rice')如果没有中间的编号，优先级相同时就会去比较第三个值。订单要是 dict 这种没法比较的值，程序就炸了。
pq = queue.PriorityQueue()
pq.put((1, {"dish": "soup"}))
try:
pq.put((1, {"dish": "rice"})) # 优先级相同，就拿 dict 和 dict 比……
except TypeError as e:
print(e)
# '<' not supported between instances of 'dict' and 'dict'还有像一摞盘子那样从最后放的开始取的 LifoQueue，以及最简单的 SimpleQueue。
plates = queue.LifoQueue() # 一摞盘子：最后放上去的先拿
for p in ("p1", "p2", "p3"):
plates.put(p)
print(plates.get(), plates.get(), plates.get())
# p3 p2 p1
simple = queue.SimpleQueue() # 最简单的订单夹：没有 maxsize、task_done、join
simple.put("#1")
print(simple.get(), hasattr(simple, "task_done"))
# #1 False| 订单夹 | 取出顺序 | 用在哪里 |
|---|---|---|
Queue | 先放的先取 | 几乎所有场合 |
PriorityQueue | 值小的先取 | VIP 优先、急活优先 |
LifoQueue | 后放的先取 | 一摞盘子、先处理最新的 |
SimpleQueue | 先放的先取 | 不需要容量限制和完成追踪的时候 |
今晚的目标是：12 份订单由三个厨师做，每份不多不少只做一次，并且确切地知道全部做完了。两个前厅员工轮流接单，订单夹有四格。
rail = queue.Queue(maxsize=4) # 订单夹有四格
served = []
def hall(orders): # 前厅员工：接单，夹上去
for n in orders:
time.sleep(0.02)
rail.put(f"#{n}")
def cook(name):
while True:
try:
order = rail.get()
except queue.ShutDown:
return
time.sleep(0.1)
served.append((name, order))
rail.task_done()
start = time.perf_counter()
halls = [threading.Thread(target=hall, args=(range(k, 13, 2),)) for k in (1, 2)]
cooks = [threading.Thread(target=cook, args=(n,)) for n in ("minho", "sua", "tak")]
for t in halls + cooks: t.start()
for t in halls: t.join() # 接单结束后
rail.shutdown() # 关掉订单夹
rail.join() # 等夹上的订单全部做完
for t in cooks: t.join()
orders = [o for _, o in served]
print(len(orders), len(set(orders)), f"{time.perf_counter() - start:.2f}s")
# 12 12 0.44s（时间只是大约）做出来的菜 12 份，互不相同的订单也是 12 份。没有一单做了两次，也没有一单漏掉。老卓也干了自己那份活。而这段代码里，连一行 Lock 都没有。
主厨轻轻敲了敲订单夹。“现在前厅和厨房不用互相喊话了。可下周五要是来一笔 30 份的团体订单呢？你打算招 30 个厨师？”
营业日志 — 第六晚。装上订单夹（Queue）之后，同一单做两次的事再没发生。关掉订单夹（shutdown），用 join 确认所有订单都做完后打烊。
cook 改成只有 #5 这一单会抛 ValueError。rail.join() 还会放行吗？为了不至于永远等下去，用 waiter = threading.Thread(target=rail.join, daemon=True) 把它放到后台，waiter.join(timeout=1) 之后打印 waiter.is_alive() 确认一下，再用 try/finally 修好它。PriorityQueue，让偶数号订单当作 VIP 先做。shutdown() 在 PriorityQueue 上也一样能用。主厨的配方卡
生产者-消费者：放的一方 rail.put(x)，取的一方 rail.get()。队列替你上了锁。
全做完了吗？每做完一件就 rail.task_done()，等待的一方用 rail.join()。
Queue(maxsize=n)：满了放的一方就得等（背压）。timeout=、put_nowait/get_nowait、queue.Full/queue.Empty。
收工：按厨师人数放哨兵 None，或者从 3.13 起用 rail.shutdown() → queue.ShutDown。immediate=True 会扔掉剩下的订单。
PriorityQueue（(优先级, 编号, 订单)）、LifoQueue（一摞盘子）、SimpleQueue（最简单）。
订阅新课时和新课程通知
本课程新增课时或推出后续课程时，我们会发邮件通知你。
问答
关于本课有任何疑问都可以在这里留言。所有语言版本页面的提问会汇集在一起。