分离协程生成与执行时，如何确保所有协程正确关闭？

我想手动控制协程执行顺序，所以选择直接处理协程而非创建任务（任务会自动被事件循环调度），但遇到了一个奇怪的Runtime警告问题：

无警告的代码示例 #

async def nested_func():
    pass


async def outer_nest():
    await nested_func()


if __name__ == '__main__':
    outer_coroutine = outer_nest()
    outer_coroutine.close()

触发警告的代码示例 #

async def passed_func():
    pass


async def caller(passed_coroutine):
    await passed_coroutine


if __name__ == '__main__':
    passed_coroutine = passed_func()
    caller_coroutine = caller(passed_coroutine)
    caller_coroutine.close()

我自己推测的原因是：调用协程函数时只会解析参数，不会执行函数体。第一种场景里，nested_func的协程要到outer_nest执行时才会被创建；但第二种场景中passed_coroutine已经被提前创建，可caller的函数体从未执行，导致这个协程被遗留，所以触发了警告。

现在想问：是不是只能通过维护一个awaitable列表，在父协程关闭时手动调用它们的close方法？比如做个管理器来处理？

你的推测完全正确，两种场景的核心差异就是协程对象的创建时机：

• 第一种情况里，nested_func()是在outer_nest的函数体内被调用的，而你直接关闭了outer_coroutine，这个协程根本没开始执行，所以nested_func的协程对象从未被创建，自然不会有遗留问题。
• 第二种情况里，passed_coroutine是在外部提前创建好的，它已经是一个存在的协程对象，但因为caller_coroutine被直接关闭，await passed_coroutine这行代码根本没机会执行，这个协程就成了“孤儿”，Python的垃圾回收机制检测到未被等待的协程，就会抛出Runtime警告。

至于解决方案，不是只能用维护列表的方式，还有更灵活的处理思路：

1. 手动关闭传入的协程 #

这是最直接的方式，在关闭父协程之前，主动调用传入协程的close()方法，避免警告。修改主函数如下：

if __name__ == '__main__':
    passed_coroutine = passed_func()
    caller_coroutine = caller(passed_coroutine)
    try:
        caller_coroutine.close()
    finally:
        passed_coroutine.close()

2. 封装协程管理器 #

如果需要处理大量这类场景，可以写一个简单的管理器类，负责跟踪所有传入的协程对象，在父协程关闭时自动遍历关闭所有子协程：

class CoroutineManager:
    def __init__(self):
        self._coros = []
    
    def track(self, coro):
        self._coros.append(coro)
        return coro
    
    def close_all(self):
        for coro in self._coros:
            coro.close()
        self._coros.clear()

# 使用示例
if __name__ == '__main__':
    manager = CoroutineManager()
    passed_coroutine = manager.track(passed_func())
    caller_coroutine = caller(passed_coroutine)
    caller_coroutine.close()
    manager.close_all()

3. 利用上下文管理器 #

把协程的创建和关闭逻辑封装到上下文管理器里，确保无论是否发生异常，协程都能被正确关闭：

from contextlib import contextmanager

@contextmanager
def sync_managed_coro(coro_func):
    coro = coro_func()
    try:
        yield coro
    finally:
        coro.close()

# 使用示例
if __name__ == '__main__':
    with sync_managed_coro(passed_func) as passed_coroutine:
        caller_coroutine = caller(passed_coroutine)
        caller_coroutine.close()

总结一下，最简单的临时解决方式就是手动关闭传入的协程；如果场景复杂，再考虑用管理器或上下文管理器来批量处理，不用局限于维护列表这一种思路。

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