今晚的订单
#10 · 第十晚
危机
团体预订，六张桌。3 号桌因为过敏问题退了套餐，整个团体说干脆改天一起来。可另外五桌的锅还在咕嘟咕嘟地煮着。
目标
一桌失败，其余的菜马上停下；而且任何一道菜都不许超过规定的时间。
深夜食堂 同时
晚上十点半，六桌团体客人同时落座。小露接了单递进厨房，厨房同时开做六套菜。十分钟后，3 号桌有人举手：花生过敏，刚才忘了说。团体的负责人说：“那今天我们就一起先走了。”
敏浩没听见，还在给 5 号桌煮汤。秀雅一把关掉他的火：“从来没定过‘一桌退了，大家都停’这条规矩啊。”
尹瑞妍主厨按下计时器。“一起开始的事，就得一起结束。今晚的活，就是把这条约定写进代码。”
先用第 9 课的 asyncio.gather 来处理团体预订。2 号桌 0.1 秒就失败了。
import asyncio
async def table(n, sec, fail=False):
await asyncio.sleep(sec)
if fail:
raise ValueError(f"T{n} allergy")
print(f"T{n} served")
async def main():
try:
await asyncio.gather(table(1, 0.3), table(2, 0.1, fail=True), table(3, 0.5))
except ValueError as e:
print("caught:", e)
await asyncio.sleep(0.6) # 让店再多开一会儿
asyncio.run(main())
# caught: T2 allergy
# T1 served
# T3 servedgather 一碰到第一个异常就立刻往上抛，所以 caught: 最先打印出来。served。gather(return_exceptions=True) 只是把异常装进结果列表里，并不会叫停其余的任务。想要“一个失败，全部停下”，得换别的工具。
asyncio.TaskGroup（3.11）把任务当作一组来管理。这种成组管理的做法，叫作结构化并发（structured concurrency）。
import asyncio
async def table(n, sec):
await asyncio.sleep(sec)
return f"T{n}"
async def main():
async with asyncio.TaskGroup() as tg:
tasks = [tg.create_task(table(n, 0.1 * n)) for n in range(1, 4)]
# 走到这里，三个任务都已经结束了
print([t.result() for t in tasks])
asyncio.run(main())
# ['T1', 'T2', 'T3']TaskGroup 做出三条承诺。
async with 块时，在块里创建的任务全都已经结束。不用再一个个 await。ExceptionGroup 往上抛。tg.create_task 会替你攥着任务，所以第 9 课说的“丢了引用、任务就没了”也不用担心。下面是 3 号桌失败的团体预订。
import asyncio
async def table(n, sec, fail=False):
try:
await asyncio.sleep(sec)
if fail:
raise ValueError(f"T{n} allergy")
print(f"T{n} served")
except asyncio.CancelledError:
print(f"T{n} cancelled")
raise # 收到的取消要原样抛回去
async def main():
try:
async with asyncio.TaskGroup() as tg:
tg.create_task(table(1, 0.05))
tg.create_task(table(2, 0.3))
tg.create_task(table(3, 0.1, fail=True))
tg.create_task(table(4, 0.5))
except* ValueError as eg:
print("group failed:", eg.exceptions)
await asyncio.sleep(0.6)
asyncio.run(main())
# T1 served
# T2 cancelled （T2 和 T4 的先后每次运行都不同）
# T4 cancelled
# group failed: (ValueError('T3 allergy'),)served。和 gather 不同，没有哪口锅在后台偷偷煮着。await 那里抛出 asyncio.CancelledError。所以可以用 except 接住它做收尾。收到的取消一定要再抛回去（10-5）。下面的时间线在服务器上跑的是真正的 TaskGroup。桌号越大，菜就做得越久一点（T1 0.3 秒，T2 0.345 秒，…… T6 约 0.53 秒）。被选为“会失败的那桌”，会在自己的菜快做完时失败。
厨房时间线
订单
6
会失败的那桌
无
时间限制（秒）
关
烹饪
等待
已取消
失败
几个任务也可能同时失败。所以 TaskGroup 抛出的不是一个异常，而是一组异常。
import asyncio
async def boom(exc, sec):
await asyncio.sleep(sec)
raise exc
async def main():
try:
async with asyncio.TaskGroup() as tg:
tg.create_task(boom(ValueError("T2 allergy"), 0.1))
tg.create_task(boom(KeyError("T5"), 0.1))
except Exception as e:
print(type(e).__name__, "|", e)
print(e.exceptions)
asyncio.run(main())
# ExceptionGroup | unhandled errors in a TaskGroup (2 sub-exceptions)
# (ValueError('T2 allergy'), KeyError('T5'))ExceptionGroup（3.11）是一种把异常列表装在 .exceptions 里的异常。要在这一组里按类型分别接住，就用 except*。
eg = ExceptionGroup("kitchen", [ValueError("T2 allergy"), KeyError("T5"), ValueError("T6 salt")])
try:
raise eg
except* ValueError as g:
print("ValueError:", [str(e) for e in g.exceptions])
except* KeyError as g:
print("KeyError:", g.exceptions)
# ValueError: ['T2 allergy', 'T6 salt']
# KeyError: (KeyError('T5'),)except 只会执行一个。except* 则会执行所有匹配的块。上面两个块都跑了。g 永远是 ExceptionGroup，是只装着匹配异常的一个小组。except* 都没匹配上的异常，会组成剩下的一组，继续往上抛。eg = ExceptionGroup("kitchen", [ValueError("T2"), KeyError("T5")])
try:
try:
raise eg
except* ValueError:
print("handled ValueError")
except ExceptionGroup as rest:
print("left:", rest.exceptions)
print(isinstance(eg, Exception), eg.subgroup(ValueError).exceptions)
# handled ValueError
# left: (KeyError('T5'),)
# True (ValueError('T2'),)except* 有两条规矩。在 3.14 里，两条都会在运行前就被当作语法错误拦下来。
src1 = '''
def f():
try:
pass
except* ValueError:
return 1
'''
src2 = '''
try:
pass
except ValueError:
pass
except* KeyError:
pass
'''
for src in (src1, src2):
try:
compile(src, "<menu>", "exec")
except SyntaxError as e:
print("SyntaxError:", e.msg)
# SyntaxError: 'break', 'continue' and 'return' cannot appear in an except* block
# SyntaxError: cannot have both 'except' and 'except*' on the same 'try'except* 块里不能用 return、break、continue。几个块要依次执行，中途跳出去就乱套了。把结果存进变量，到 try 外面再返回。try 里不能混用 except 和 except*。两种都需要，就把 try 拆成两层。本课程实验台的引擎也是这么写的。主厨的提醒
在 except* 里想 return 的时候，把结果记进变量，到 try 外面再交出去。厨房也一样：事故报告得全部收齐了才下班，可不能听完第一份就关门走人。
今晚的第二条规矩是“任何一道菜都不许超时”。asyncio.timeout()（3.11）给整个块加上时间限制。
import asyncio, time
async def stew(sec):
await asyncio.sleep(sec)
return "stew"
async def main():
start = time.perf_counter()
try:
async with asyncio.timeout(0.2):
await stew(1.0)
except TimeoutError:
print(f"timeout after {time.perf_counter() - start:.2f}s")
try:
print(await asyncio.wait_for(stew(0.1), timeout=0.2))
print(await asyncio.wait_for(stew(1.0), timeout=0.2))
except TimeoutError:
print("wait_for: timeout")
print(asyncio.TimeoutError is TimeoutError)
asyncio.run(main())
# timeout after 0.20s
# stew
# wait_for: timeout
# TrueTimeoutError。从 3.11 开始，asyncio.TimeoutError 和内置的 TimeoutError 就是同一个东西。asyncio.wait_for(协程, timeout=) 是给单个协程加限制的老办法。需要直接拿到结果时很方便。asyncio.timeout 能把好几个 await 包进一个块里，比如“从点单到上菜，加起来不超过 20 分钟”。限制也可以不按“几秒之后”，而按“几点几分”来设，还可以在进入块之后再改。
import asyncio, time
async def main():
loop = asyncio.get_running_loop()
start = time.perf_counter()
deadline = loop.time() + 0.3 # 从现在起 0.3 秒后这个“时刻”
try:
async with asyncio.timeout_at(deadline):
await asyncio.sleep(1)
except TimeoutError:
print(f"timeout_at: {time.perf_counter() - start:.1f}s")
start = time.perf_counter()
try:
async with asyncio.timeout(None) as cm: # 一开始不设限
cm.reschedule(loop.time() + 0.2) # 之后再定为 0.2 秒
print(cm.when() is not None)
await asyncio.sleep(1)
except TimeoutError:
print(f"reschedule: {time.perf_counter() - start:.1f}s, expired={cm.expired()}")
asyncio.run(main())
# timeout_at: 0.3s
# True
# reschedule: 0.2s, expired=Truetimeout_at(时刻) 接收的是以 loop.time() 为基准的绝对时刻。好几个步骤共用同一个“截止时刻”时很好用。timeout(None) 不带限制地开始。之后可以用 cm.reschedule(时刻) 设定或推迟限制。TaskGroup 也好，时间限制也好，说到底干的是同一件事：给任务发取消。要亲手发，就用 task.cancel()。
import asyncio
async def braise():
print("stove on")
try:
await asyncio.sleep(10)
finally:
print("stove off") # 就算被取消，火也要关
async def main():
task = asyncio.create_task(braise())
await asyncio.sleep(0.1)
task.cancel()
try:
await task
except asyncio.CancelledError:
print("cancelled:", task.cancelled())
asyncio.run(main())
# stove on
# stove off
# cancelled: Truecancel() 不会当场把任务杀掉。它只是让任务在下一个停着的 await 处抛出 CancelledError。try/finally。不管是取消、成功还是失败，都一定会执行。CancelledError 继承的不是 Exception，而是 BaseException。这样设计，是为了不让 except Exception: 不小心把取消吞掉。要是把取消吞掉，会怎样？
import asyncio, time
async def stubborn():
try:
await asyncio.sleep(0.5)
except asyncio.CancelledError:
pass # 把取消吞了
await asyncio.sleep(0.5) # 然后接着做菜
return "late"
async def main():
start = time.perf_counter()
try:
async with asyncio.timeout(0.1):
print(await stubborn())
except TimeoutError:
print("timeout")
print(f"{time.perf_counter() - start:.1f}s")
asyncio.run(main())
# late
# 0.6s明明设了 0.1 秒的限制，却花了 0.6 秒，TimeoutError 也没出现。时间限制靠取消来干活，可这个取消被协程吃掉了。TaskGroup 的“取消其余任务”也会这样垮掉。所以接到 CancelledError 后，只做收尾，然后再抛回去（raise）。
主厨的提醒
接到取消，就关火、擦砧板，然后直接举手。一想着“再煮一会儿就好了”接着煮，厨房的规矩就全塌了。敏浩，说的就是你。
有些活，比如保存收据，中途绝不能断。asyncio.shield() 能挡住外面来的取消，不让它传到里面的任务。
import asyncio
async def save_receipt():
await asyncio.sleep(0.3)
print("receipt saved")
return "ok"
async def main():
inner = asyncio.create_task(save_receipt())
try:
async with asyncio.timeout(0.1):
await asyncio.shield(inner)
except TimeoutError:
print("waiter gave up")
print(await inner) # 里面的任务一直在跑
asyncio.run(main())
# waiter gave up
# receipt saved
# ok等着的一方 0.1 秒就放弃了，里面的任务却跑完了全程。得一直攥着里面那个任务的引用，之后才能拿到结果。
Task.cancelling() 返回该任务收到的取消请求数，Task.uncancel() 把这个数减一。这是 asyncio.timeout 和 TaskGroup 在内部分辨“这个取消是不是我发的”时用的工具，自己直接用到的机会很少。
import asyncio
async def worker():
try:
await asyncio.sleep(1)
except asyncio.CancelledError:
me = asyncio.current_task()
print("cancelling:", me.cancelling())
me.uncancel()
print("after uncancel:", me.cancelling())
return "rescued"
async def main():
t = asyncio.create_task(worker())
await asyncio.sleep(0.05)
t.cancel()
print(await t)
asyncio.run(main())
# cancelling: 1
# after uncancel: 0
# rescued把今晚的两条规矩用在同一个厨房里。每桌限时 0.6 秒，六桌绑进同一个 TaskGroup。
import asyncio, time
async def table(n, sec, allergy, stopped):
try:
async with asyncio.timeout(0.6): # 任何一道菜都不超过 0.6 秒
await asyncio.sleep(sec)
if n == allergy:
raise ValueError(f"T{n} allergy")
return f"T{n}"
except asyncio.CancelledError:
stopped.append(f"T{n}") # 关火，举手
raise
async def service(menu, allergy=None):
start, stopped = time.perf_counter(), []
try:
async with asyncio.TaskGroup() as tg:
tasks = [tg.create_task(table(n, s, allergy, stopped)) for n, s in menu.items()]
print("served:", [t.result() for t in tasks])
except* ValueError as eg:
print("cancelled by:", [str(e) for e in eg.exceptions], "stopped:", sorted(stopped))
except* TimeoutError as eg:
print("too slow:", len(eg.exceptions))
print(f"{time.perf_counter() - start:.1f}s")
MENU = {1: 0.30, 2: 0.35, 3: 0.20, 4: 0.40, 5: 0.45, 6: 0.50}
asyncio.run(service(MENU))
asyncio.run(service(MENU, allergy=3))
asyncio.run(service({**MENU, 6: 0.9}))
# served: ['T1', 'T2', 'T3', 'T4', 'T5', 'T6']
# 0.5s
# cancelled by: ['T3 allergy'] stopped: ['T1', 'T2', 'T4', 'T5', 'T6']
# 0.2s
# too slow: 1
# 0.6sTimeoutError 就会打包抛上来，其余的同样停下。快打烊时，靠窗的客人吃完了那碗面，合上了小本子。结账时，她对主厨简短地说了一句：“下周五我再来。”
主厨点点头，回头看向副厨。“取消也好，时间限制也好，说到底都发生在 await 那里。可厨房角落那台旧烤箱，根本不懂什么叫 await。小露要是站在它跟前干等，前厅会变成什么样？”
营业日志 — 第十晚。六桌团体绑进同一个 TaskGroup，3 号桌一退，其余的锅同时关火。每桌都设了时间限制，并把“不许吞掉取消”的规矩贴在了厨房墙上。
gather 例子改成 TaskGroup，让 2 号桌失败时，1 号和 3 号打印出 cancelled。table 的 raise 删掉，在那个位置换成 await asyncio.sleep(0.3)（无视取消，接着煮），然后用 allergy=3 运行。看看耗时怎么变了，写下哪里出了问题。主厨的配方卡
gather 在一个失败时不会叫停其余任务。必须一起结束的活，用 TaskGroup（3.11）绑在一起。
async with asyncio.TaskGroup() as tg: tg.create_task(...) — 出了块就全部结束；一个失败，其余取消。
失败会汇总成 ExceptionGroup。用 except* ValueError 按类型接（不许 return，不许和 except 混用）。
时间限制：async with asyncio.timeout(秒) 或 wait_for(协程, timeout=)。超时抛内置 TimeoutError。
取消会在 await 处以 CancelledError 的形式到来。用 finally 收尾，收到的取消要再 raise。
不能被打断的活用 asyncio.shield(task)。cancelling()/uncancel() 是写工具的人用的机关。
订阅新课时和新课程通知
本课程新增课时或推出后续课程时，我们会发邮件通知你。
问答
关于本课有任何疑问都可以在这里留言。所有语言版本页面的提问会汇集在一起。