如何在Python中使用anyio编写兼容asyncio和trio的代码？-Python教程-PHP中文网

能，anyio通过统一api屏蔽底层差异，仅当代码严格使用其抽象接口（如anyio.sleep、anyio.create_task_group）且避免任何asyncio/trio原生调用时，才可真正跨backend兼容。

如何在python中使用anyio编写兼容asyncio和trio的代码？

anyio 能否真正同时兼容 asyncio 和 trio？

能，但前提是代码只调用 anyio 提供的抽象 API，不直接依赖 asyncio 或 trio 的原生函数。一旦用了 asyncio.create_task()、trio.sleep() 或 asyncio.run() 这类底层调用，就锁死了运行时。anyio 的兼容性不是“自动转换”，而是“统一接口屏蔽差异”——它靠运行时选择 backend 实现，而非在代码里做条件判断。

如何写出可切换 backend 的 anyio 代码？

核心是只使用 anyio 模块导出的顶层函数和类，避免任何对 asyncio 或 trio 的显式 import。常见错误包括：误用 asyncio.sleep()（应改用 anyio.sleep()）、手动创建 asyncio.Queue（应改用 anyio.create_memory_object_stream() 或 anyio.Queue）。

anyio.run() 是唯一启动点，必须指定 backend="asyncio" 或 backend="trio"，不能省略
所有 I/O 操作走 anyio.open_file()、anyio.connect_tcp()、anyio.serve_tcp() 等封装函数
并发控制用 anyio.create_task_group()（对应 trio 的 nursery / asyncio 的 TaskGroup），不用 asyncio.gather() 或 trio.nursery.start_soon()
取消与超时统一用 anyio.fail_after()、anyio.move_on_after()，不混用 asyncio.timeout() 或 trio.move_on_after()

为什么用 anyio.Queue 却在 asyncio 下报 AttributeError？

因为 anyio.Queue 在 trio backend 下基于 trio.MemorySendChannel 实现，在 asyncio 下则包装了 asyncio.Queue，但它的 API 并非完全对齐。最常踩的坑是：试图访问 .qsize() 或 .empty() —— 这些方法在 trio backend 下不存在，anyio 未做跨 backend 补全。正确做法是只用标准接口：put()、get()、close()，以及通过 anyio.create_memory_object_stream() 替代需要大小检查的场景。

示例对比：

python-script-generator

快速生成专业的 Python 脚本和应用代码。一键创建完整项目结构，支持CLI、API、爬虫、Bot、Django等多种项目类型，包含完整的项目结构、配置文件、依赖管理、测试、README和文档。

下载

# ❌ 错误：qsize() 在 trio 下会 AttributeError
q = anyio.Queue(10)
await q.put("item")
if q.qsize() > 5:  # 运行时报错
    ...
<h1>✅ 正确：用流 + 统计变量或仅依赖阻塞语义</h1><p>send_stream, receive_stream = anyio.create_memory_object_stream()</p><h1>后续只用 send_stream.send() / receive_stream.receive()</h1><p></p>

测试时如何确保两种 backend 都跑通？

不能只跑一次 anyio.run(main, backend="asyncio") 就认为 OK。必须显式为每个 backend 启动独立进程或子解释器（因 backend 在首次 anyio.run() 后即全局锁定）。推荐方式是在 pytest 中用参数化 fixture：

用 @pytest.mark.anyio（需装 pytest-anyio）自动覆盖两个 backend
若手动测试，分别执行：python -m pytest test.py --anyio-backend=asyncio 和 --anyio-backend=trio
特别注意资源清理：anyio.Path 的 .unlink()、anyio.open_file() 的 close() 必须在 finally 或 async with 中完成，否则 trio 下可能 hang 住