今晚的订单
#11 · 第十一晚
危机
二手买来的旧烤箱，门一关上，不烤完就不松开把手。小露站在它跟前的这段时间，整个前厅都停了。
目标
烤箱烤着的时候，其他桌照样 0.3 秒上菜。
深夜食堂 同时
晚上十一点。3 号桌点了焗菜。焗菜只能用主厨二手淘来的那台旧烤箱烤。这台烤箱门一关，不烤完把手就锁死，还得有人一直握着把手站在那儿。
小露站到了烤箱前。就在这一刻，前厅停了。要水的 1 号桌，要结账的 5 号桌，都只能干等小露回来。
老卓啧了一声：“站在烤箱跟前不叫干活，那叫站着。”主厨点点头：“小露不能等。等这种事，交给别人。”
asyncio 的事件循环跑在一个线程上。协程得在 await 处让出来，循环才能转去别的任务。可要是调用了不懂 await 的函数，比如 time.sleep 或者老式的同步库，循环就只能僵在那里，直到这个函数结束。这种调用叫作阻塞调用。
import asyncio, time
start = time.perf_counter()
async def lu_rounds(): # 小露每 0.1 秒绕前厅一圈
for _ in range(5):
print(f"round {time.perf_counter() - start:.1f}s")
await asyncio.sleep(0.1)
def old_oven(): # 不懂 await 的老函数
time.sleep(0.35)
return "bread"
async def bake():
await asyncio.sleep(0.15)
print("oven", old_oven()) # 整个循环在这里停住 0.35 秒
async def main():
await asyncio.gather(lu_rounds(), bake())
asyncio.run(main())
# round 0.0s
# round 0.1s
# oven bread
# round 0.5s
# round 0.6s
# round 0.7stime.sleep、requests.get 这类同步网络库、读大文件、繁重的计算。在下面的时间线里，有一桌慢的要用旧烤箱（time.sleep，0.6 秒）。其他桌每次 0.1 秒、分几段用 asyncio.sleep 等。
厨房时间线
订单
6
慢的那桌
1
交给 to_thread
烹饪
等待
已取消
失败
程序一大，就很难知道是哪个协程霸占了循环。asyncio 的调试模式会把一次执行太久的回调和任务记进日志。
import asyncio, logging, time
logging.basicConfig(level=logging.WARNING, format="%(name)s %(levelname)s: %(message)s")
def old_oven():
time.sleep(0.3)
async def bake():
old_oven()
async def main():
await asyncio.create_task(bake(), name="T3-oven")
asyncio.run(main(), debug=True)
# 例：asyncio WARNING: Executing <Task finished name='T3-oven' coro=<bake() done, defined at ...oven.py:8>
# result=None created at ...\asyncio\tasks.py:395> took 0.301 secondsasyncio.run(main(), debug=True) 打开。也可以用环境变量 PYTHONASYNCIODEBUG=1 或 python -X dev 打开。T3-oven）、协程定义所在的文件和行号，以及霸占的时间（took 0.301 seconds）。给任务加上 name=，找元凶就容易多了。asyncio 日志记录器的 WARNING 级别输出。路径因电脑而异。判定标准是 loop.slow_callback_duration，默认 0.1 秒。想抓得更灵敏，就把它调小。
import asyncio, time
async def main():
loop = asyncio.get_running_loop()
print(loop.get_debug(), loop.slow_callback_duration)
loop.slow_callback_duration = 0.05 # 超过 0.05 秒就告诉我
time.sleep(0.07)
asyncio.run(main(), debug=True)
# True 0.1
# 例：Executing <Task finished name='Task-1' coro=<main() done, defined at ...> result=None created at ...> took 0.070 seconds调试模式还会告诉你：没被 await 就丢掉的协程是在哪里创建的；从别的线程乱碰循环时，它也会报错。开发时打开，上线时关掉（会变慢）。
旧烤箱修不好，那就另派一个人站到它跟前。asyncio.to_thread()（3.9）把阻塞函数放到另一个线程里跑，再给循环交回一个可以 await 的对象。
import asyncio, time
start = time.perf_counter()
async def lu_rounds():
for _ in range(5):
print(f"round {time.perf_counter() - start:.1f}s")
await asyncio.sleep(0.1)
def old_oven(item, *, minutes):
time.sleep(0.35)
return f"{item} {minutes}m"
async def bake():
await asyncio.sleep(0.15)
bread = await asyncio.to_thread(old_oven, "bread", minutes=40)
print("oven", bread)
async def main():
await asyncio.gather(lu_rounds(), bake())
asyncio.run(main())
# round 0.0s
# round 0.1s
# round 0.2s
# round 0.3s
# round 0.4s
# oven bread 40mawait 收结果。to_thread(函数, *args, **kwargs) 会把参数原样传过去。它在第 7 课的线程池（循环默认的 executor）里运行。繁重的计算，就送去分店厨房，也就是进程池。用 asyncio.get_running_loop() 拿到正在运行的循环，再用 loop.run_in_executor(executor, 函数, *args)。因为用到进程，要存成文件再运行。
import asyncio
from concurrent.futures import ProcessPoolExecutor
def knead(n): # 手不停的活（CPU）
return sum(i * i for i in range(n))
async def main():
loop = asyncio.get_running_loop()
with ProcessPoolExecutor(max_workers=2) as pool:
doughs = await asyncio.gather(
loop.run_in_executor(pool, knead, 2_000_000),
loop.run_in_executor(pool, knead, 3_000_000),
)
print(doughs)
if __name__ == "__main__":
asyncio.run(main())
# [2666664666667000000, 8999995500000500000]run_in_executor 的第一个参数传 None，就用默认的线程池。to_thread 正是把这件事变方便的函数。run_in_executor 不接受关键字参数。需要的话，用 functools.partial(knead, n=...) 包一层再传。if __name__ == "__main__":，传过去的函数和值都必须能被 pickle。主厨的提醒
在 async def 里调一个名字上既没 async 也没 await 的函数时，先停一下问自己：“这个要花很久吗？”哪怕只要 0.1 秒，也该交给 to_thread。别让小露站到烤箱跟前。
第 4 课和第 6 课的那些工具，都另有 asyncio 版。名字一样，但等待时不是让线程停下，而是用 await 让出来。
| threading | asyncio | 用法 |
|---|---|---|
threading.Lock | asyncio.Lock | async with lock: |
threading.Semaphore | asyncio.Semaphore | async with sem: |
threading.Event | asyncio.Event | await event.wait() |
queue.Queue | asyncio.Queue | await q.put(x)、await q.get() |
信号量最常用来“限制同时连接数”。要给供货商打五通电话，可对方同时只接得了两通。
import asyncio, time
start = time.perf_counter()
supplier = asyncio.Semaphore(2) # 供货商的电话同时最多两通
async def call(n):
async with supplier:
print(f"call #{n} at {time.perf_counter() - start:.1f}s")
await asyncio.sleep(0.2)
async def main():
await asyncio.gather(*(call(n) for n in range(1, 6)))
asyncio.run(main())
# call #1 at 0.0s
# call #2 at 0.0s
# call #3 at 0.2s
# call #4 at 0.2s
# call #5 at 0.4sasyncio.Queue 和第 6 课的订单夹是一个样子。maxsize、task_done()、join() 也都有。
import asyncio
async def hall(q):
for n in range(1, 6):
await q.put(f"#{n}") # 订单夹满了就在这里等
print("hall done")
async def cook(name, q, done):
while True:
ticket = await q.get()
await asyncio.sleep(0.1)
done.append((name, ticket))
q.task_done()
async def main():
q = asyncio.Queue(maxsize=2)
done = []
cooks = [asyncio.create_task(cook(c, q, done)) for c in ("A", "B")]
await hall(q)
await q.join() # 等夹上的订单全部 task_done
for c in cooks:
c.cancel()
print(len(done), sorted(t for _, t in done))
asyncio.run(main())
# hall done
# 5 ['#1', '#2', '#3', '#4', '#5']循环只有一个线程，看起来不会有竞争条件。可只要读和写之间夹着 await，第 3 课的黑板事故就会原样重演。
import asyncio
async def main():
board = {"count": 0}
async def cook():
for _ in range(100):
v = board["count"]
await asyncio.sleep(0) # 在这里切到别的任务
board["count"] = v + 1
await asyncio.gather(*(cook() for _ in range(4)))
print(board["count"])
asyncio.run(main())
# 100本该是 400 的黑板，写着 100。四个任务读到同一个值，让出去，再写回同一个值。用 asyncio.Lock 把这一段包起来，再用 asyncio.Event 等“开张！”的信号，就是这样：
import asyncio
async def main():
lock = asyncio.Lock()
opened = asyncio.Event()
board = {"count": 0}
async def cook():
await opened.wait() # 等到“开张！”
for _ in range(100):
async with lock: # 读和写之间有 await，就得上锁
v = board["count"]
await asyncio.sleep(0)
board["count"] = v + 1
tasks = [asyncio.create_task(cook()) for _ in range(4)]
await asyncio.sleep(0.1)
opened.set()
await asyncio.gather(*tasks)
print(board["count"])
asyncio.run(main())
# 400两个世界的工具不能混用。在协程里等 threading.Lock，整个循环都会停住，攥着锁的那个任务也就没机会回来把锁打开了。
import asyncio, threading, time
lock = threading.Lock() # 在协程里用线程用的锁会怎样？
start = time.perf_counter()
async def cook(n):
if not lock.acquire(timeout=0.2): # 等的这段时间，整个循环都停着
print(f"#{n} gave up at {time.perf_counter() - start:.1f}s")
return
try:
await asyncio.sleep(0.1) # 锁着就让出去了
print(f"#{n} done")
finally:
lock.release()
async def main():
await asyncio.gather(cook(1), cook(2))
asyncio.run(main())
# #2 gave up at 0.2s
# #1 done要是没有 timeout，#2 会永远等下去，#1 也永远醒不过来。反过来，asyncio 的工具不是线程安全的，不能从别的线程往 asyncio.Queue 里放东西。从线程把活交给循环的办法，在 11-6 里讲。
with 和 for 也有需要等待的版本。
async with：进入和离开时都需要 await 的资源（__aenter__、__aexit__）。比如锁、信号量、网络连接。async for：取下一个值得等的循环（__aiter__、__anext__）。比如一张张到来的订单、流式响应。async def 里写 yield，就成了能用 async for 遍历的迭代器。import asyncio
from contextlib import asynccontextmanager
async def tickets(n): # 异步生成器：async def + yield
for i in range(1, n + 1):
await asyncio.sleep(0.05) # 等下一张订单到来
yield f"#{i}"
@asynccontextmanager
async def stove(name): # 用 async with 来用的灶台
await asyncio.sleep(0.01)
print(f"{name} on")
try:
yield name
finally:
print(f"{name} off")
async def main():
async with stove("S1"):
async for t in tickets(3):
print("cook", t)
print([t async for t in tickets(2)])
asyncio.run(main())
# S1 on
# cook #1
# cook #2
# cook #3
# S1 off
# ['#1', '#2']contextlib.asynccontextmanager 只用一个 yield，就能做出给 async with 用的资源。列表推导式里也能写 async for（仅限协程内部）。
第 7 课的 as_completed 也有 asyncio 版。从 3.13 起可以用 async for 遍历，这时它会按完成顺序，交回当初放进去的那些任务本身。
import asyncio
async def dish(n, sec):
await asyncio.sleep(sec)
return f"#{n}"
async def main():
tasks = [asyncio.create_task(dish(n, s)) for n, s in [(1, 0.3), (2, 0.1), (3, 0.2)]]
async for t in asyncio.as_completed(tasks): # 3.13 起
print(t in tasks, await t)
asyncio.run(main())
# True #2
# True #3
# True #1照老办法写成 for f in asyncio.as_completed(tasks): await f 也行，但那时拿到的 f 不是原来的任务，而是新对象。想知道是哪一单做完了，async for 更方便。
假设接预订电话的员工在另一个线程里干活。这个线程要往小露的日程里插一件事，就用 asyncio.run_coroutine_threadsafe(协程, loop)。拿回来的是第 7 课的 concurrent.futures.Future，所以在线程里可以用 result(timeout=) 等着。
import asyncio
async def seat(table):
await asyncio.sleep(0.05)
return f"{table} seated"
def phone_line(loop): # 循环外面的线程（预订电话）
fut = asyncio.run_coroutine_threadsafe(seat("T7"), loop)
print(fut.result(timeout=1)) # concurrent.futures.Future
async def main():
loop = asyncio.get_running_loop()
await asyncio.to_thread(phone_line, loop)
asyncio.run(main())
# T7 seatedeager 任务（3.12）：一般来说，create_task 只是把任务排进日程，轮到它之前并不执行。用了 asyncio.eager_task_factory，任务一创建就立刻执行到第一个 await。如果协程根本不需要 await 就能结束，它连日程都不用进，当场就做完了。
import asyncio
async def greet(n):
print(f"greet T{n}") # 第一个 await 之前
await asyncio.sleep(0)
return n
async def main(eager):
if eager:
asyncio.get_running_loop().set_task_factory(asyncio.eager_task_factory)
t = asyncio.create_task(greet(1))
print("created, done =", t.done())
await t
asyncio.run(main(False))
asyncio.run(main(True))
# created, done = False
# greet T1
# greet T1
# created, done = False查看正在跑的任务（3.14）：只要知道正在运行的 asyncio 程序的进程号（PID），就能在另一个终端里用 python -m asyncio ps <PID> 看任务列表，用 python -m asyncio pstree <PID> 以树的形式看谁在等谁。比如 hall 任务正通过 TaskGroup 等着 T1、T2，输出会是这样（省略了部分列和路径）：
task name coroutine stack awaiter name
Task-1 main
hall TaskGroup._aexit -> TaskGroup.__aexit__ -> hall Task-1
T1 sleep -> table hall
T2 sleep -> table hall这是在“前厅停了，可不知道小露站在哪儿”的时候用的工具。给任务加上 name=，列表会好读得多。
只有 3 号桌的焗菜放进旧烤箱，烤箱跟前派一个线程守着。用一个开关（fix）来对比。
import asyncio, time
def old_oven(sec): # 动不了的老库
time.sleep(sec)
return "gratin"
async def table(n, start, fix):
if n == 3:
dish = await asyncio.to_thread(old_oven, 0.6) if fix else old_oven(0.6)
else:
await asyncio.sleep(0.3)
dish = "noodle"
return f"T{n} {dish} {time.perf_counter() - start:.1f}s"
async def service(fix):
start = time.perf_counter()
async with asyncio.TaskGroup() as tg:
tasks = [tg.create_task(table(n, start, fix)) for n in range(1, 7)]
print([t.result() for t in tasks])
asyncio.run(service(fix=False))
asyncio.run(service(fix=True))
# ['T1 noodle 0.6s', 'T2 noodle 0.6s', 'T3 gratin 0.6s', 'T4 noodle 0.9s', 'T5 noodle 0.9s', 'T6 noodle 0.9s']
# ['T1 noodle 0.3s', 'T2 noodle 0.3s', 'T3 gratin 0.6s', 'T4 noodle 0.3s', 'T5 noodle 0.3s', 'T6 noodle 0.3s']to_thread，面 0.3 秒出锅，焗菜在烤箱时间 0.6 秒出炉。这就是今晚的目标。打烊后，主厨用手指一张张点过厨房墙上的配方卡：线程、锁、队列、池、进程、循环、TaskGroup、to_thread。“工具攒了这么多。可客人推门进来的时候，先拿哪一件，要怎么定？”
营业日志 — 第十一晚。用调试模式查出前厅每次停 0.6 秒，是因为在等旧烤箱。烤箱交给 to_thread 的线程，供货商的电话用 asyncio.Semaphore 限制在两通以内。
asyncio.run(main(), debug=True) 运行 11-1 的第一个例子，看看警告里说哪个任务霸占了几秒。asyncio.Semaphore(3)。先算一算 call #5 会在第几秒开始，再运行验证。主厨的配方卡
阻塞调用（time.sleep、同步库）会让循环上的所有任务停下。
揪出元凶：asyncio.run(main(), debug=True)、loop.slow_callback_duration（默认 0.1 秒）。
等待型的阻塞代码用 await asyncio.to_thread(f, *args)。繁重计算用 loop.run_in_executor(ProcessPoolExecutor(), f, x)。
asyncio 版工具：Lock、Semaphore、Event、Queue — 用 async with / await。不要和 threading 版混用。
async with、async for、异步生成器（async def + yield）。3.13 起可写 async for t in asyncio.as_completed(tasks)。
从线程交给循环：run_coroutine_threadsafe(coro, loop)。查看任务：python -m asyncio ps/pstree <PID>（3.14）。
订阅新课时和新课程通知
本课程新增课时或推出后续课程时，我们会发邮件通知你。
问答
关于本课有任何疑问都可以在这里留言。所有语言版本页面的提问会汇集在一起。