如何回收Python后台asyncio任务?类实例GC时任务是否同步回收?
类实例被垃圾回收时,asyncio任务会同步回收吗?
当MyClass的实例被垃圾回收时,你创建的这些asyncio任务不会同步被回收或停止,必须主动处理,原因和解决方案如下:
核心原因
asyncio.create_task()创建的任务会被当前事件循环持有强引用,哪怕你的类实例被回收,事件循环依然会持续调度这些任务运行。任务的生命周期不绑定到创建它的实例上,只要事件循环在运行,无限循环的my_task_body就会一直占用资源。- 你提到的“全局弱引用”是误解,最新稳定版Python中,事件循环对任务的引用是强引用,不存在自动回收任务的机制。
为什么必须主动停止?
如果不主动处理,这些后台任务会一直运行下去,导致内存泄漏,甚至拖慢整个程序的执行效率——毕竟每个任务都在占用CPU时间片和内存资源。
解决方案
方案1:用__del__终结器清理任务
虽然__del__的执行时机不确定,但可以用来兜底取消任务:
import asyncio class MyClass: def __init__(self): self.__background_tasks = [] for i in range(3): task = asyncio.create_task(self.my_task_body()) self.__background_tasks.append(task) async def my_task_body(self): while True: # 示例工作逻辑,避免忙等 await asyncio.sleep(1) def __del__(self): # 取消所有未完成的任务 for task in self.__background_tasks: if not task.done(): task.cancel() # 处理任务取消或执行中的异常 try: task.result() except (asyncio.CancelledError, Exception): pass
方案2:用异步上下文管理器(推荐)
上下文管理器能更优雅地控制任务的生命周期,避免依赖__del__的不确定性:
import asyncio class MyClass: def __init__(self): self.__background_tasks = [] async def __aenter__(self): # 进入上下文时创建任务 for i in range(3): task = asyncio.create_task(self.my_task_body()) self.__background_tasks.append(task) return self async def __aexit__(self, exc_type, exc_val, exc_tb): # 退出上下文时取消并等待任务结束 for task in self.__background_tasks: if not task.done(): task.cancel() # 等待所有任务完成,允许返回异常避免崩溃 await asyncio.gather(*self.__background_tasks, return_exceptions=True) async def my_task_body(self): while True: # 示例工作逻辑 await asyncio.sleep(1) # 使用方式 async def main(): async with MyClass() as obj: # 在这里使用MyClass实例 await asyncio.sleep(5) asyncio.run(main())
总结
- 实例被GC不会自动停止asyncio任务,必须主动取消
- 优先使用异步上下文管理器来管理任务生命周期,比
__del__更可靠 - 取消任务后,建议等待任务完成,确保资源被正确清理
内容的提问来源于stack exchange,提问作者zapta
相关产品推荐
相关产品推荐

