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 或 BaseExceptionGroup 聚合传播，未被 except* 匹配的部分继续向外抛出：

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 anyio.sleep(0)）；任意 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 的异步客户端；取消等待不会强行终止已运行的工作线程。

官方参考

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