AsyncIO：coroutine、await 与事件循环

目标：能够预测 coroutine、Task 和顺序 await 的执行行为，并识别会阻塞事件循环的同步调用。

async def 定义 coroutine function；调用它只创建 coroutine object，不会立刻执行函数体。要推进 coroutine，需要：

• 在另一个 coroutine 中 await；
• 用 asyncio.create_task() 调度成 Task；
• 在顶层用 asyncio.run() 启动事件循环。

await 的含义不是“自动开线程”。当 awaitable 暂停等待 I/O 时，当前 Task 把控制权还给 event loop，让同一线程推进其他可运行 Task。这是协作式并发。

完整示例：顺序与并发

保存为 async_lab.py：

from __future__ import annotations

import asyncio
from time import perf_counter


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


async def sequential() -> list[str]:
    return [await fetch("a", 0.03), await fetch("b", 0.03)]


async def concurrent() -> list[str]:
    first = asyncio.create_task(fetch("a", 0.03))
    second = asyncio.create_task(fetch("b", 0.03))
    return [await first, await second]


async def main() -> None:
    start = perf_counter()
    one = await sequential()
    sequential_time = perf_counter() - start

    start = perf_counter()
    two = await concurrent()
    concurrent_time = perf_counter() - start

    assert one == two == ["A", "B"]
    assert concurrent_time < sequential_time
    print("concurrent overlaps waiting")


if __name__ == "__main__":
    asyncio.run(main())

运行：

uv run python async_lab.py

预期输出的稳定部分：顺序模式先完成 a 再开始 b；并发模式会先打印两个 start，最后打印：

concurrent overlaps waiting

不要把 0.03 秒当性能承诺。sleep 只用来观察调度关系。

阻塞事件循环

在 async function 内调用 time.sleep()、同步 HTTP client 或长 CPU 循环，不会因为外层有 async def 就变成非阻塞。它们会占住事件循环线程。

可以用 heartbeat 观察：

async def heartbeat() -> None:
    for _ in range(3):
        print("tick")
        await asyncio.sleep(0.01)

若另一个 Task 执行 time.sleep(0.1)，tick 会出现明显空洞。同步阻塞 I/O 可以用 await asyncio.to_thread(func, ...) 过渡；CPU 工作则在第 14 课选择执行模型。

引导实验：创建但不等待 coroutine

把 await sequential() 改成单独调用 sequential() 且不保存、不 await。运行结束时通常出现 RuntimeWarning: coroutine was never awaited。修复方式不是关闭 warning，而是明确：直接 await、创建并保存 Task，或把相关子任务放入 TaskGroup。

独立练习

写 fetch_with_timeout(name, delay, limit)，使用 asyncio.timeout()；delay 超过 limit 时返回 "timeout:<name>"。

async def fetch_with_timeout(name: str, delay: float, limit: float) -> str:
    try:
        async with asyncio.timeout(limit):
            return await fetch(name, delay)
    except TimeoutError:
        return f"timeout:{name}"


async def check() -> None:
    assert await fetch_with_timeout("slow", 0.05, 0.01) == "timeout:slow"


asyncio.run(check())

常见误区

• await 等于并发：连续两个 await 仍然是顺序，除非先创建 Task 或使用 TaskGroup。
• asyncio.sleep“释放 GIL”：它挂起当前 Task；GIL 是线程执行 Python bytecode 的另一层问题。
• notebook 内嵌套 asyncio.run：已有事件循环时直接在支持的环境 await main()，脚本中继续用 asyncio.run()。

参考资料

• Python 3.12: Coroutines and Tasks