异步编程 asyncio

Web 爬虫、LLM API 调用、数据库连接池都涉及 I/O 等待；asyncio 用单线程事件循环调度协程，避免线程开销。需先理解 python-fundamentals 中的异常与模块；Agent 并发请求是典型场景，见 ai-agents-essentials；/roadmaps/python。

学习目标

• 区分阻塞 I/O、多线程与 asyncio 的适用场景
• 编写 async def 协程并用 asyncio.gather 并发
• 理解 await、Task、超时与取消

最小示例

import asyncio

async def fetch_lesson(slug: str, delay: float) -> str:
    await asyncio.sleep(delay)  # 模拟网络 I/O
    return f"content of {slug}"

async def main() -> None:
    results = await asyncio.gather(
        fetch_lesson("pya-async", 0.5),
        fetch_lesson("pya-web", 0.3),
    )
    print(results)

asyncio.run(main())

超时与异常

async def with_timeout() -> None:
    try:
        await asyncio.wait_for(fetch_lesson("slow", 10), timeout=2.0)
    except asyncio.TimeoutError:
        print("请求超时")

async def shield_cancel() -> None:
    task = asyncio.create_task(fetch_lesson("important", 1))
    await asyncio.shield(task)

何时不用 async

• CPU 密集（图像处理、大矩阵）：用 multiprocessing 或交给 C 扩展
• 简单脚本：同步代码 + requests 往往足够
• 混用阻塞库：在 asyncio.to_thread() 中跑阻塞函数，避免卡住事件循环

import asyncio

def blocking_io() -> str:
    import time
    time.sleep(1)
    return "done"

async def run_blocking() -> str:
    return await asyncio.to_thread(blocking_io)

练习

• 用 gather 并发请求 3 个模拟 API，打印总耗时
• 给其中一个任务加 0.5s 超时，捕获 TimeoutError
• 说明为何在 async 函数里直接 time.sleep(1) 会阻塞整个 loop

小结

asyncio 解决 I/O 并发，不解决 CPU 并行。 下一课把 async 接到 FastAPI，构建可上线的 Web 服务。