今晚的订单
#07 · 第七晚
危机
一家公司聚餐，30 份团体订单一下子全进来了。总不能招 30 个厨师。一个个手动建线程、再把结果收进列表的代码，也早就写烦了。
目标
把 30 份订单交给六个外派厨师，1 秒内做完，做好一份就上一份。烧焦的盘子也一个不漏。
深夜食堂 同时
第七晚，十点半。电话响了，主厨放下听筒：“公司聚餐，团体，30 人份。”
敏浩一边重新系围裙一边问：“要建 30 个线程吗？”主厨摇摇头。“自己招人，管吃管住发工资，最后还得送人走，全是我们的事。今天给劳务公司打电话。人归他们管，我们只管把活扔过去。”
concurrent.futures.ThreadPoolExecutor 就是一家劳务公司：事先养着几个线程，活一来就交给空闲的线程。第 2 课里亲手做的那些事，建 Thread、start、join、准备装结果的列表，它全都替你干了。
import time
from concurrent.futures import ThreadPoolExecutor
def cook(n):
time.sleep(0.2)
return f"#{n} ready"
with ThreadPoolExecutor(max_workers=3) as agency: # 三个外派厨师
fut = agency.submit(cook, 1) # 交出活，换回一张“取餐券”（Future）
print(fut.done()) # 还在做
print(fut.result()) # 等它做完，拿到结果
print(fut.done())
# False
# #1 ready
# Truesubmit(函数, 参数...) 交出活后会立刻返回一个 Future。它不是菜本身，而是“凭它取菜的券”。print(fut) 一下，能看到 <Future at 0x... state=running> 这样的状态。fut.result() 会一直等到菜做好，然后返回函数的返回值。第 2 课里“线程不返回值”、只好往列表里塞的问题，到这里就解决了。with 块时会调用 shutdown(wait=True)：等交出去的活全部做完，再把线程收拾干净。现在交出 30 份。厨师只有五个，所以每人轮流接六份。
import threading
def cook(n):
time.sleep(0.1)
return threading.current_thread().name
start = time.perf_counter()
with ThreadPoolExecutor(max_workers=5, thread_name_prefix="agency") as agency:
futures = [agency.submit(cook, n) for n in range(1, 31)] # 30 份订单
# 出了 with 块，所有任务都已经做完了
names = {f.result() for f in futures}
print(len(futures), sorted(names), f"{time.perf_counter() - start:.2f}s")
# 30 ['agency_0', 'agency_1', 'agency_2', 'agency_3', 'agency_4'] 0.60s订单有 30 份，线程却只建了五个。30 份 0.1 秒的订单由五个人分担，用时 0.6 秒。建线程、销毁线程的开销只付一次，同一个厨师一单接一单地做下去。这就叫线程池（thread pool）。
拿结果有两种办法。这次让 #1 订单最费时间。
from concurrent.futures import as_completed
def cook(n):
time.sleep(0.1 * (5 - n)) # #1 最费时间（0.4 秒 … 0.1 秒）
return f"#{n}"
with ThreadPoolExecutor(max_workers=4) as agency:
start = time.perf_counter()
futures = [agency.submit(cook, n) for n in range(1, 5)]
for f in as_completed(futures): # 完成的顺序
print(f.result(), f"{time.perf_counter() - start:.1f}s", end=" ")
print()
start = time.perf_counter()
for dish in agency.map(cook, range(1, 5)): # 提交的顺序
print(dish, f"{time.perf_counter() - start:.1f}s", end=" ")
print()
# #4 0.1s #3 0.2s #2 0.3s #1 0.4s
# #1 0.4s #2 0.4s #3 0.4s #4 0.4sas_completed(futures) | executor.map(函数, 多个值) | |
|---|---|---|
| 返回什么 | 一个一个地返回已完成的 Future | 一个一个地返回结果值 |
| 顺序 | 完成的顺序 | 提交的顺序 |
| 第一个结果 | 最快的那个一做完就出来 | 要等 #1 做完 |
| 适合的场合 | 做好一份上一份、显示进度 | 结果要和输入一一对应、原样使用时 |
用 map 的话，#4 虽然 0.1 秒就做好了，也要等到 #1 做完的 0.4 秒才交出来，因为它得守住提交的顺序。总耗时一样，但第一盘菜上桌的时间不一样。
as_completed 只返回 Future，想知道是哪一单的结果，惯例是用 dict 把两者配好对。
with ThreadPoolExecutor(max_workers=4) as agency:
table_of = {agency.submit(cook, n): f"T{n}" for n in range(1, 5)} # Future → 桌号
for fut in as_completed(table_of):
print(table_of[fut], fut.result())
# T4 #4
# T3 #3
# T2 #2
# T1 #1下面时间线里的“外派厨师”，跑的也是真正的 ThreadPoolExecutor。这个场景特意设计成编号越靠后的订单越快做完。
厨房时间线
订单
6
厨师
3
结果返回顺序
烹饪
等待
已取消
失败
as_completed 时，返回顺序是乱的，比如 #3 → #2 → #1 → #6 → #5 → #4 → …。因为三个人同时开工的订单里，短的先做完。map 时，永远是 #1 → #2 → … → #10。as_completed 的返回顺序几乎完全倒过来，像 #8 → #7 → … → #1。因为八份订单同时开工，编号靠后的短单先做完。交出去的活里要是抛了异常，这个异常会被悄悄存进 Future 里。直到你调用 result() 的那一刻，它才在那里重新炸开。
def cook(n):
if n == 3:
raise ValueError(f"#{n} burnt") # 3 号订单烧焦了
return f"#{n}"
with ThreadPoolExecutor(max_workers=2) as agency:
futures = [agency.submit(cook, n) for n in range(1, 5)]
# 到这里为止一片安静，好像什么都没发生
for f in futures:
try:
print(f.result())
except ValueError as e: # 调用 result() 时才炸
print("error:", e)
# #1
# #2
# error: #3 burnt
# #4
print(repr(futures[2].exception()), futures[0].exception())
# ValueError('#3 burnt') Noneresult()，异常就会什么也不打印地消失。好比烧焦的盘子虽然没端给客人，却也没有任何人知道。exception() 能把异常取出来，而不让它炸开。成功的话返回 None。map 在取结果的过程中会当场抛出异常，之后的结果就拿不到了。如果一单失败了还要拿到其余的，就用 submit 加 as_completed。想在一做完就马上干点什么，可以用 add_done_callback 挂个铃铛。传进去一个函数，它接收一个已完成的 Future 作为参数。
def cook(n):
time.sleep(0.1)
return f"#{n}"
def ring_bell(fut): # 做完时被调用的函数（回调）。接收一个 Future
print("bell:", fut.result(), threading.current_thread().name)
with ThreadPoolExecutor(max_workers=2, thread_name_prefix="agency") as agency:
for n in (1, 2):
agency.submit(cook, n).add_done_callback(ring_bell)
# 例：
# bell: #1 agency_0
# bell: #2 agency_1回调是在干完活的那个外派厨师的线程里被调用的（如果挂到一个已经完成的 Future 上，就在挂铃铛的那个人的线程里立刻调用）。所以如果在回调里修改共享数据，第 3 课的锁规矩照样适用。回调里抛的异常只会记进日志，然后被忽略。
主厨的提醒
只 submit 不调 result 的代码，是我在现场见得最多的错。异常就在取餐券里睡着了。交出去的活，要么用 result 拿回来，要么用 exception 查一查，必须二选一。
cancel() 只能取消还没开始的活。已经下锅的菜，可就收不回来了。
with ThreadPoolExecutor(max_workers=1) as agency: # 一个厨师
futures = [agency.submit(time.sleep, 0.2) for _ in range(3)]
time.sleep(0.05) # 给第一单下锅的时间
print([f.cancel() for f in futures]) # 已经开始的收不回来
print([f.cancelled() for f in futures], [f.done() for f in futures])
# [False, True, True]
# [False, True, True] [False, True, True]cancel() 取消成功返回 True；已经在做或者做完了，返回 False。done() 在做完或已取消时为 True。第一单还在做，所以是 False。要一次性结束和劳务公司的合同，就用 shutdown(cancel_futures=True)。排着队的活全部取消，只等已经开始的活做完。
agency = ThreadPoolExecutor(max_workers=2)
futures = [agency.submit(time.sleep, 0.2) for _ in range(6)]
time.sleep(0.05)
agency.shutdown(cancel_futures=True) # 排队的订单取消，锅上的订单做完
print(sum(f.cancelled() for f in futures), sum(not f.cancelled() for f in futures))
# 4 2一堆活里只要有一个做完就想行动，用 wait。
from concurrent.futures import FIRST_COMPLETED, wait
def cook(sec):
time.sleep(sec)
return sec
with ThreadPoolExecutor(max_workers=3) as agency:
futures = [agency.submit(cook, s) for s in (0.3, 0.1, 0.2)]
done, pending = wait(futures, return_when=FIRST_COMPLETED) # 只要有一个做完
print([f.result() for f in done], len(pending))
done, pending = wait(futures, timeout=0.15) # 默认是 ALL_COMPLETED
print(len(done), len(pending))
# [0.1] 2
# 2 1wait 返回两个集合：做完的和没做完的。return_when 可选 FIRST_COMPLETED、FIRST_EXCEPTION、ALL_COMPLETED（默认值）；过了 timeout，没全做完也会返回。它不会抛异常，所以得自己检查 pending 是不是空的。
concurrent.futures 的好处在于合同是同一份。submit、map、Future、as_completed 都不用动，只换公司名就行。
from concurrent.futures import InterpreterPoolExecutor, ProcessPoolExecutor, ThreadPoolExecutor
def knead(n):
return sum(range(n))
if __name__ == "__main__": # 进程池必须有这一句（第 8 课）
for Agency in (ThreadPoolExecutor, ProcessPoolExecutor, InterpreterPoolExecutor):
with Agency(max_workers=2) as agency:
print(Agency.__name__, list(agency.map(knead, [10, 100, 1000])))
# ThreadPoolExecutor [45, 4950, 499500]
# ProcessPoolExecutor [45, 4950, 499500]
# InterpreterPoolExecutor [45, 4950, 499500]| 公司 | 厨师 | 适合的活 | 课 |
|---|---|---|---|
ThreadPoolExecutor | 同一个厨房里的线程 | 要等的活（I/O） | 7 |
ProcessPoolExecutor | 分店厨房（进程） | 手不停的活（CPU） | 8 |
InterpreterPoolExecutor | 同一栋楼里隔开的小厨房（子解释器），3.14 起 | 手不停的活，比进程更轻 | 8 |
这个例子别在交互式 shell 里跑，要存成 .py 文件再运行。进程池为什么非要 if __name__ == "__main__":，明晚再说。
省略 max_workers 时，线程池会叫来 min(32, os.process_cpu_count() + 4) 个人（3.12 及以前是 os.cpu_count()）。在这台 12 核电脑上就是 16 个。要等的活需要比核数更多的线程，所以加 4；又怕太多，所以封顶 32。实际的服务里，最好考虑对方服务器能扛住多少，自己定这个数。
今晚的目标是：把 30 份订单交给六个外派厨师，1 秒内做好一份上一份，烧焦的盘子也一个不漏。每份订单要 0.1~0.2 秒，13 号订单会烧焦。
def cook(n):
time.sleep(0.1 + 0.05 * (n % 3)) # 0.1~0.2 秒
if n == 13:
raise ValueError(f"#{n} burnt")
return f"#{n}"
served, burnt = [], []
start = time.perf_counter()
with ThreadPoolExecutor(max_workers=6, thread_name_prefix="agency") as agency:
futures = {agency.submit(cook, n): n for n in range(1, 31)}
for fut in as_completed(futures): # 做好一份上一份
try:
served.append(fut.result())
except ValueError:
burnt.append(futures[fut])
print(len(served), burnt, f"{time.perf_counter() - start:.2f}s")
# 29 [13] 0.80s（时间只是大约）29 盘菜做好一份上一份，烧焦的 13 号进了重做清单。用时 0.8 秒。没有建线程的代码，也没有事先准备好装结果的列表。
靠窗的客人今天也是一个人，面前一碗面。团体订单的盘子一盘盘端出去的时候，她看了一眼手表，在小本子上简短地记了点什么。
主厨送走外派厨师们，说道：“明天要揉的面堆成山。叫四个外派厨师来揉，能快四倍吗？”
营业日志 — 第七晚。30 份团体订单交给 6 个外派厨师（ThreadPoolExecutor），0.8 秒做完。13 号烧焦了，但在 result() 那里当场抓到，重新做了一份。
as_completed 换成 agency.map(cook, range(1, 31))。到 13 号时会发生什么？能拿到几盘菜？wait(..., return_when=FIRST_COMPLETED) 和 shutdown(cancel_futures=True) 写一段代码：30 份订单里只要第一盘做好的，其余全部取消。已经开始做的菜有几份？主厨的配方卡
with ThreadPoolExecutor(max_workers=n) as ex: fut = ex.submit(f, x); fut.result()
Future 是取餐券：done()、result()、exception()、cancel()、add_done_callback(fn)。
as_completed(futures) 按完成顺序，ex.map(f, xs) 按提交顺序。
任务里的异常在调用 result() 时才炸。不调用，它就悄悄消失。
cancel() 只对还没开始的任务有效。一次性取消用 ex.shutdown(cancel_futures=True)。只等一个用 wait(fs, return_when=FIRST_COMPLETED)。
同一份合同：ProcessPoolExecutor（第 8 课）、InterpreterPoolExecutor（3.14）。默认线程数是 min(32, os.process_cpu_count() + 4)。
订阅新课时和新课程通知
本课程新增课时或推出后续课程时，我们会发邮件通知你。
问答
关于本课有任何疑问都可以在这里留言。所有语言版本页面的提问会汇集在一起。