Python anyio 的跨生态适配案例

不能直接替代，但可间接兼容：fastapi 硬依赖 asyncio 运行时，anyio 需显式指定 backend=”asyncio”才能安全调用 asyncio 库（如 aiomysql），且仅限完全可控的子逻辑中使用。

anyio 能不能直接替代 asyncio 在 FastAPI 里用

不能直接替代，但可以间接兼容。FastAPI 底层依赖 asyncio 的事件循环和任务调度机制，而 anyio 是独立运行时，不共享 asyncio 的 loop 或 TaskGroup 实例。

常见错误现象：RuntimeError: no current Event loop 或 Task was destroyed but it is pending，尤其在测试或嵌套调用中出现——本质是两个运行时混用导致上下文丢失。

• 如果你用 anyio.run() 启动主逻辑，FastAPI 的 uvicorn 就无法接管事件循环
• FastAPI 的依赖注入、中间件、后台任务都硬编码适配 asyncio，绕不开
• 真正能用 anyio 的地方，仅限于你完全控制的子逻辑：比如在某个路由 handler 内部用 anyio.create_task_group() 并发调用多个 I/O 操作

在 anyio 环境里安全调用 asyncio-only 的库（如 aiomysql）

多数 asyncio 生态库（aiomysql、aiohttp、asyncpg）内部直接依赖 asyncio.get_event_loop()，在 anyio 的 trio 或 curio 后端下会直接崩溃。

解决路径只有一条：强制使用 asyncio 后端，并确保整个调用链不跨后端切换。

立即学习“Python免费学习笔记（深入）”；

• 启动时显式指定：anyio.run(main, backend="asyncio")
• 避免在同一个 anyio 任务里混用 asyncio.sleep() 和 anyio.sleep()，它们行为不一致
• asyncio.to_thread() 可以桥接同步阻塞调用，但 asyncio.to_thread() 本身只能在 asyncio 后端下工作

示例：安全调用 aiomysql

import anyio import aiomysql <p>async def fetch_data(): pool = await aiomysql.create_pool(host='127.0.0.1', port=3306, ...) async with pool.acquire() as conn: async with conn.cursor() as cur: await cur.execute("SELECT 1") return await cur.fetchone()</p><h1>必须这样跑</h1><p>anyio.run(fetch_data, backend="asyncio")

anyio.TaskGroup 和 asyncio.TaskGroup 的关键差异

两者名字像，但设计目标完全不同：anyio.TaskGroup 是跨后端统一的结构化并发原语；asyncio.TaskGroup（python 3.11+）只是对 asyncio.create_task() 的语法糖封装，且仍绑定 asyncio 运行时。

容易踩的坑：

• 别试图把 asyncio.TaskGroup 实例传给 anyio 函数——类型不兼容，运行时报 TypeError
• anyio.TaskGroup 的 cancel_scope 是可嵌套、可手动触发的，而 asyncio.TaskGroup 的取消由异常传播隐式触发，不可控
• 若你在 anyio 中用 asyncio.create_task()，任务不会被 anyio.TaskGroup 捕获，可能泄漏

跨生态日志与异常传播的实际约束

日志记录器（如 structlog）本身无运行时偏好，但一旦涉及异步上下文绑定（例如 request ID 注入），就暴露底层差异。

典型问题：在 trio 后端下用 contextvars 存储请求 ID，部分 asyncio 库（如旧版 httpx）会丢失该变量——因为 contextvars 的上下文隔离粒度取决于运行时实现，anyio 不保证跨后端透传。

• 不要依赖 contextvars.ContextVar 在 anyio 和第三方 asyncio 库之间传递状态
• 异常堆栈里混着 anyio 和 asyncio 的帧？说明你无意中触发了后端切换，检查是否漏写了 backend="asyncio"
• 调试时优先用 anyio.get_current_task() 而非 asyncio.current_task()，后者在非 asyncio 后端下抛 RuntimeError

跨生态不是“换一个 import 就能跑”，而是明确边界在哪、哪些模块必须锁死后端、哪些调用必须加适配胶水——这些边界往往藏在你没注意的第三方库内部。