一个循环,一个生命周期
asyncio.run() 创建新事件循环、运行协程并关闭循环。一个进程里调用两次没问题,前提是第一次运行的任何对象都不再存活。异步客户端、会话和后台任务往往会存活,并在之后尝试使用已关闭的循环。
async def main():
async with httpx.AsyncClient() as client:
await fetch_all(client)
asyncio.run(main())
客户端应在 main 内创建,而不是在模块导入时。返回前要 await 或取消后台任务。不要保留循环引用并跨多次 asyncio.run 复用。
测试需要匹配作用域
在一个循环中创建客户端、在另一个循环中关闭的测试夹具会产生此错误。使用 pytest-asyncio 的异步夹具,让事件循环作用域与资源一致,并在同一循环中关闭客户端。同一进程跑两次测试可以暴露泄漏的循环状态。
关闭待处理工作
关闭时取消剩余任务并 await 其取消。asyncio.all_tasks() 显示仍在运行的任务。待处理时被销毁的任务可能记录警告并丢失工作。目标是干净关闭循环,而不是用异常处理器掩盖症状。
用调试模式验证
使用 PYTHONASYNCIODEBUG=1 和 warnings-as-errors 运行。确认没有残留待处理任务,并反复运行测试套件。如果错误仍在,找出在循环外创建的对象;那就是生命周期不匹配。