含asyncio Task的对象超出作用域后未及时销毁的问题及优化管理方案咨询

这个问题我之前也碰到过，核心原因是循环引用搞的鬼，咱们一步步拆解来看：

当你创建asyncio.Task的时候，Task对象会持有它所运行的协程的引用；而你的heartbeat协程是BackgroundTask的实例方法，它隐式地持有了对BackgroundTask实例（也就是bar）的引用——这就形成了一个循环引用链：bar → task → heartbeat协程 → bar。

CPython的垃圾回收器虽然能处理循环引用，但它的回收时机是不确定的，不会在对象刚出作用域就立刻触发，得等到事件循环结束或者GC主动运行的时候才会清理这个循环引用，这就是为什么bar要等到main结束才被销毁。

关于反复调用foo的风险 #

如果foo被频繁调用，这些BackgroundTask实例确实会在内存里累积，直到GC触发回收。要是创建速度远快于GC回收速度，内存占用会持续升高，长期运行下来可能引发性能问题甚至内存不足，必须重视。

解决方案 #

方案一：显式打破循环引用

在取消任务后，手动把task引用置为None，切断循环链，让对象能被及时回收：

import asyncio

class BackgroundTask:
    def __init__(self):
        self.task = asyncio.create_task(self.heartbeat())
        
    async def heartbeat(self):
        while True:
            await asyncio.sleep(1)
            print("^")
    
    def __del__(self):
        print("bar deleted")

async def foo():
    bar = BackgroundTask()
    await asyncio.sleep(2)
    bar.task.cancel()
    bar.task = None  # 显式切断循环引用

async def main():
    await foo()
    print("foo has ended")
    await asyncio.sleep(2)
    print("main ended")

asyncio.run(main())

修改后运行，bar deleted会在foo has ended之前打印，符合预期。

方案二：用异步上下文管理器自动管理（更优雅）

给BackgroundTask实现异步上下文管理器接口，让它支持async with语法，在退出上下文时自动取消任务并清理引用，不用手动写重复代码：

import asyncio

class BackgroundTask:
    def __init__(self):
        self.task = None
        
    async def __aenter__(self):
        self.task = asyncio.create_task(self.heartbeat())
        return self
        
    async def heartbeat(self):
        while True:
            await asyncio.sleep(1)
            print("^")
            
    async def __aexit__(self, exc_type, exc, tb):
        if self.task:
            self.task.cancel()
            try:
                await self.task  # 等待任务响应取消，避免控制台警告
            except asyncio.CancelledError:
                pass
            self.task = None  # 清理引用
            
    def __del__(self):
        print("bar deleted")

async def foo():
    async with BackgroundTask() as bar:
        await asyncio.sleep(2)

async def main():
    await foo()
    print("foo has ended")
    await asyncio.sleep(2)
    print("main ended")

asyncio.run(main())

这种方式更符合Python的“资源获取即初始化”（RAII）思想，代码可读性和可维护性更好。

MicroPython专属注意事项

MicroPython的垃圾回收机制比CPython简单，而且没有__del__方法，对循环引用的处理能力很弱（甚至完全不处理），所以必须更主动地管理资源：

• 优先用上面的异步上下文管理器方案，或者给类加一个显式的close方法，在不用的时候主动调用：

import asyncio

class BackgroundTask:
    def __init__(self):
        self.task = asyncio.create_task(self.heartbeat())
        
    async def heartbeat(self):
        while True:
            await asyncio.sleep(1)
            print("^")
            
    async def close(self):
        self.task.cancel()
        try:
            await self.task
        except asyncio.CancelledError:
            pass
        self.task = None  # 必须显式切断引用

async def foo():
    bar = BackgroundTask()
    await asyncio.sleep(2)
    await bar.close()

async def main():
    await foo()
    print("foo has ended")
    await asyncio.sleep(2)
    print("main ended")

asyncio.run(main())

在MicroPython里，一定要显式清理Task引用，不然对象会一直占用内存，直到设备重启。

备注：内容来源于stack exchange，提问作者sandyscott