AnyIO 异步编程入门指南（Python）

AnyIO 提供兼容 asyncio/trio 的高级异步 API，跨后端运行需依赖库同时兼容。本文讲解任务组、结构化取消、同步 with 超时作用域与异常组处理。

观测云编辑部 2026-09-29 3 分钟阅读更新于 2026-09-29

直接回答：AnyIO 是一个 Python 异步编程库，提供统一的高级 API，让你的异步代码可以在 asyncio、trio 两个后端上运行，但依赖库也必须兼容所选后端，同时带来任务组、结构化并发、优雅取消等现代特性。

为什么需要 AnyIO

Python 异步生态长期存在"选边站"问题：用 asyncio 写的库和用 trio 写的库互不兼容。AnyIO 在两者之上抽象出统一接口——你写的代码面对 AnyIO API，底层跑哪个后端由部署方决定。FastAPI、Starlette 等现代框架的异步层正是建立在 AnyIO 之上。

安装与起步

python3 -m venv venv && source venv/bin/activate
pip install anyio

一个最小例子：

import anyio

async def main():
    print("开始")
    await anyio.sleep(1)
    print("一秒后")

anyio.run(main)   # 默认 asyncio 后端；anyio.run(main, backend="trio") 切换

使用 Trio 前安装 pip install 'anyio[trio]'。仅使用 AnyIO API 的代码较容易切换后端；asyncio 专用依赖不能直接在 Trio 上运行。下文 except* 语法要求 Python 3.11+。

任务组：结构化并发

AnyIO 不用裸的 create_task，而是用任务组（task group）保证"所有子任务结束，父作用域才退出"：

import anyio

async def worker(name, delay):
    await anyio.sleep(delay)
    print(f"{name} 完成")

async def main():
    async with anyio.create_task_group() as tg:
        tg.start_soon(worker, "任务A", 2)
        tg.start_soon(worker, "任务B", 1)

anyio.run(main)

这就是结构化并发：任务不会"飘"出作用域，异常会聚合传播，便于通过 finally 或上下文管理器清理资源——比手工管理一堆 task 句柄可靠得多。

错误处理与取消

任务组内任一任务抛异常，其余任务会被自动取消，异常以 ExceptionGroup 形式聚合抛出：

async def main():
    try:
        async with anyio.create_task_group() as tg:
            tg.start_soon(risky_operation)
            tg.start_soon(other_operation)
    except* ValueError as eg:
        print("捕获到 ValueError 组:", eg.exceptions)

子任务要响应取消时，在关键点 await 任意异步操作即可（取消是协作式的）。

超时控制

async def fetch_with_timeout():
    with anyio.move_on_after(5):      # 最多 5 秒，超时静默跳过
        return await slow_api()
    return "超时降级结果"

async def strict():
    with anyio.fail_after(5):   # 超时抛 TimeoutError
        await slow_api()

move_on_after 适合"超时就算了"的场景，fail_after 适合"超时必须处理"的场景——比手工 wait_for 语义更清晰。

常见问题（FAQ）

Q：项目直接用 asyncio，有必要换 AnyIO 吗？
A：写库（尤其是想同时服务 asyncio/trio 用户的库）强烈推荐；写应用且确定只用 asyncio，直接用 asyncio 也没问题。理解 AnyIO 的任务组模型对写健壮的 asyncio 代码同样有帮助。

Q：AnyIO 的性能开销大吗？
A：需在相同依赖、并发与取消行为下测量实际负载；不能统一认定开销可忽略。

Q：协程里能用 requests 这类同步库吗？
A：直接调用会阻塞事件循环。用 anyio.to_thread.run_sync() 把同步调用丢到线程池执行，或改用 httpx 这类原生异步客户端。

官方参考

本文基于官方文档整理，未进行运行时或性能测试。示例中的业务函数、数据模型和部署地址需结合项目补全；局部片段不等同于完整生产应用。