Python3.9中如何正确await asyncio任务组并避免程序永久运行

故障根因 #

• 程序永久挂起无法退出的核心原因：在main协程内部调用asyncio.all_tasks()时，返回的任务集合会包含当前正在运行的main协程对应的Task对象——调用该方法时，main协程仍处于未完成的活跃状态，属于事件循环追踪的活跃任务范畴，自然会被纳入返回集合。后续对全量集合执行asyncio.gather()等待，本质是让main协程等待自身执行完成，形成无终止条件的自等待死锁。
• 能力缺口：asyncio.all_tasks()没有内置“排除当前调用上下文对应任务”的过滤逻辑，无法直接满足“收集自定义任务、排除主运行任务”的需求。

修复实现 #

1. 实现排除主任务的全量任务收集 #

要实现all_tasks()的等价过滤版本，只需在获取全量任务集合后，剔除当前正在运行的任务对象即可。Python 3.9中可通过asyncio.current_task()获取当前上下文正在执行的Task对象，直接做集合差集就能得到目标任务列表。
对应代码修改片段：

async def main():
    asyncio.create_task(f())
    asyncio.create_task(f())
    # 排除当前运行的主任务
    current_task = asyncio.current_task()
    pending = {task for task in asyncio.all_tasks() if task is not current_task}
    group = asyncio.gather(*pending, return_exceptions=True)
    await group

修改后，等待集合中不再包含main自身，待两个自定义创建的f()任务执行完成后，main协程即可正常退出，asyncio.run()会自动完成事件循环收尾，程序正常退出。这种过滤逻辑适用于需要收集事件循环内除当前主运行任务外所有活跃任务的场景，完全匹配开发者想要的all_tasks()排除主任务的等价能力。

2. 更稳妥的工程化替代方案 #

全局遍历收集全量任务的写法存在混入无关任务的风险，更可控的实现是在创建子任务时手动维护任务集合，从逻辑上避免混入主任务或其他无关任务：

async def main():
    # 手动维护专属任务集合
    custom_tasks = set()
    custom_tasks.add(asyncio.create_task(f()))
    custom_tasks.add(asyncio.create_task(f()))
    # 仅等待自定义创建的任务
    group = asyncio.gather(*custom_tasks, return_exceptions=True)
    await group

该方案的优势：

• 从设计上规避主任务被纳入等待集合的问题，完全没有自等待死锁的可能
• 不会误纳入事件循环中其他位置创建的无关任务，逻辑边界清晰
• 完全兼容Python 3.9的运行环境

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