asyncio 异步编程：协程、await、gather

理解 asyncio 的协作式并发：惰性协程、挂起点、gather、TaskGroup，以及阻塞调用带来的陷阱。

10 分钟 · 3 题

asyncio 在单个线程上实现了协作式并发：一个事件循环负责调度各个协程，而协程会在每个 await 处主动交出控制权。有一个常被误解的根本要点：调用一个 async def 函数不会运行任何东西。它只是构造出一个静止的协程对象，只有当它被 await 等待，或者作为任务被调度到事件循环上时，函数体才会向前推进。

import asyncio

async def greet() -> str:
    return "hello"

coro = greet()
print(type(coro).__name__)

async def main() -> None:
    print(await coro)

asyncio.run(main())

await 标记了一个挂起点：协程在此冻结，事件循环收回控制权，转而推进另一个就绪的任务。在两个 await 之间，你的代码不会被打断地运行 - 对大多数数据结构来说都不需要加锁。要（在 I/O 的意义上）并行启动多个协程，就把它们交给 asyncio.gather：三次一秒钟的等待总共只花一秒，而不是三秒。

import asyncio
import time

async def fetch(name: str, delay: float) -> str:
    await asyncio.sleep(delay)
    return f"{name} done"

async def main() -> None:
    start = time.perf_counter()
    results = await asyncio.gather(
        fetch("a", 1.0),
        fetch("b", 1.0),
        fetch("c", 1.0),
    )
    elapsed = time.perf_counter() - start
    print(results)
    print(f"total: {elapsed:.1f} s")

asyncio.run(main())

默认情况下，gather 会把第一个抛出的异常向外传播，却让其他任务继续在后台运行 - 这很少是你想要的。加上 return_exceptions=True 之后，异常会作为结果被返回，留待之后梳理。从 Python 3.11 起，请优先使用 asyncio.TaskGroup：一旦出现第一个错误，所有兄弟任务都会被干净地取消，而多个异常会被打包进一个 ExceptionGroup 向上冒泡，用 except* 来捕获。

import asyncio
import time

def blocking_io() -> str:
    time.sleep(1)
    return "report written"

async def main() -> None:
    results = await asyncio.gather(
        asyncio.to_thread(blocking_io),
        asyncio.sleep(1, result="tick"),
    )
    print(results)

asyncio.run(main())

知识检测

确认你已牢记本课的重点内容。

1. 单独写下 fetch('a', 1.0) 这个调用，既不 await 也不 create_task，会产生什么？
   • 一个静止的协程对象：函数体里没有一行代码会运行
   • 协程被立刻同步执行
   • 一个被自动调度到事件循环上的任务
   • 一个 SyntaxError，因为 await 是必需的
2. 通过 asyncio.gather 一起启动三个 asyncio.sleep(1)：总耗时是多少？
   • 大约 3 秒，因为只有一个线程在运行代码
   • 不知道核心数就无法预测
   • 大约 1 秒：这些等待彼此重叠，事件循环在任务之间来回切换
3. 为什么在协程里禁止使用 time.sleep(2)？
   • 它在异步上下文中会抛出异常
   • 它会阻塞事件循环所在的线程：在这 2 秒里没有任何其他任务能推进
   • 它只是比 asyncio.sleep 精度更低
   • 它占用两倍的内存