Python Asyncio如何基于上下文管理器实现Eventloop

核心结论 #

完全可以自行实现符合PEP 343规范的asyncio事件循环上下文管理器，无需依赖官方内置的loop.run()方法；只要完整覆盖官方run()方法内置的全链路资源处理逻辑，就可以完全避免内存泄漏问题。

相关背景说明 #

• 不选择loop.run()实现、转而使用上下文管理器封装事件循环的诉求具备合理性。Python官方文档多次明确提示：如果不调用标准.run()方法启动事件循环，开发者必须自行完成全链路资源回收，否则会出现内存泄漏，最容易被遗漏的场景就是异步生成器的关闭逻辑。
• Python官方开发团队此前收到过多次「内置上下文管理器版事件循环」的功能诉求，统一回应为「我们不需要该功能」，始终没有在标准库中提供对应官方实现。
• 依据PEP 343（即with语句的官方规范）给出的适用场景示例，上下文管理器完全适配asyncio低层API的使用需求：

它可以确定性关闭所有带有close方法的对象，无论是文件、生成器还是其他类型资源；即使无法提前确定对象是否需要关闭（例如接收任意可迭代对象的函数场景），上下文管理器的使用也是安全的。

实现时的必做校验项 #

自行实现上下文管理器版事件循环时，不能只做事件循环的创建和基础关闭，必须把官方loop.run()内置的以下清理逻辑全部补全，才能达到和标准实现一致的资源安全等级：

• 进入上下文阶段：完成新事件循环初始化，将其设置为当前线程的默认事件循环
• 退出上下文阶段，按固定顺序执行清理：
  • 取消所有尚未执行完成的待处理任务
  • 运行所有已取消任务的收尾回调，避免任务引用残留
  • 触发异步生成器关闭逻辑，回收未耗尽的异步生成器资源
  • 关闭事件循环绑定的默认执行器线程池，避免线程资源泄漏
  • 最终调用loop.close()回收事件循环本身占用的资源
  • 取消当前线程绑定的默认事件循环引用，避免跨上下文误用

可直接复用的最简实现 #

import asyncio
from contextlib import contextmanager

@contextmanager
def managed_event_loop():
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    try:
        yield loop
    finally:
        try:
            # 批量取消所有残留任务
            pending_tasks = asyncio.all_tasks(loop)
            for task in pending_tasks:
                task.cancel()
            # 执行任务收尾，等待取消操作完成
            loop.run_until_complete(
                asyncio.gather(*pending_tasks, return_exceptions=True)
            )
            # 回收异步生成器资源
            loop.run_until_complete(loop.shutdown_asyncgens())
            # 回收默认线程池执行器资源
            loop.run_until_complete(loop.shutdown_default_executor())
        finally:
            # 清除线程默认循环绑定，关闭循环
            asyncio.set_event_loop(None)
            loop.close()

# 使用示例
if __name__ == "__main__":
    async def test_coro():
        await asyncio.sleep(0.1)
        print("上下文管理器封装的事件循环运行正常")

    with managed_event_loop() as loop:
        loop.run_until_complete(test_coro())

内容的提问来源于stack exchange，提问作者Thingamabobs