Python中如何取消持续运行的异步父任务及子任务？

问题根源 #

1. 父任务未终止循环：runner函数在捕获CancelledError后没有跳出while True循环，导致取消信号被“吞掉”，父任务继续执行下一轮循环，不断创建新子任务，陷入无限循环。
2. 子任务未响应取消：unit_of_work的无限循环没有处理取消信号，即使被调用cancel()，也会在await asyncio.sleep()后继续执行循环。
3. 任务容器累积冗余任务：container持续添加所有创建过的子任务，每次gather都会等待所有历史任务，不仅降低效率，还会导致取消逻辑混乱。

修复方案 #

针对以上问题，修改代码如下：

import asyncio


async def runner(container):
    print(f"Executing Runner")
    iter = 0
    try:
        while True:
            await asyncio.sleep(1)
            print("Creating task")

            new_tasks = [
                asyncio.create_task(unit_of_work(), name=f"unit_of_work-{iter}"),
                asyncio.create_task(unit_of_work(), name=f"unit_of_work-{iter+1}"),
            ]
            iter += 2
            container.extend(new_tasks)

            # 清理已完成的任务，只保留活跃任务
            container[:] = [t for t in container if not t.done()]
            await asyncio.gather(*container)
    except asyncio.CancelledError:
        print("Runner任务被取消，开始终止所有子任务")
        # 取消所有剩余活跃子任务
        for task in container:
            if not task.done():
                task.cancel(f"Cancelling: {task.get_name()}")
        # 等待所有子任务完成取消流程，忽略异常避免阻塞
        await asyncio.gather(*container, return_exceptions=True)
        print("Runner及所有子任务已终止")


async def unit_of_work():
    task = asyncio.current_task()
    name = task.get_name() if task else "unknown"
    step = 0
    try:
        while True:
            await asyncio.sleep(0.5)
            print(f"{name} doing some work, step - {step}")
            step += 1
    except asyncio.CancelledError:
        print(f"{name} 收到取消信号，终止工作")
        raise  # 重新抛出取消错误，让父任务感知子任务状态


async def task_create():
    cancellables = []
    t = asyncio.create_task(runner(cancellables), name="runner")
    return t, cancellables


async def main():
    t, cancellables = await task_create()

    await asyncio.sleep(2)

    # 只需取消父任务，由父任务统一处理子任务终止
    t.cancel(f"Cancelling {t.get_name()}")
    try:
        await t
    except asyncio.CancelledError:
        pass


if __name__ == "__main__":
    asyncio.run(main())

关键修改说明 #

• 父任务循环终止：将while True包裹在try-except块中，捕获CancelledError后主动清理所有子任务，函数自然结束，彻底终止循环。
• 子任务响应取消：在unit_of_work中添加取消信号捕获逻辑，打印日志后重新抛出错误，确保父任务能正确感知子任务的取消状态。
• 优化任务管理：每次gather前清理容器内已完成的任务，避免累积无效任务；取消父任务时统一处理所有子任务，简化外部调用逻辑。
• 简化外部操作：外部只需触发父任务的取消，无需单独处理子任务，减少重复操作和逻辑混乱。

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