如何取消asyncio TaskGroup中的所有任务？

你想要在coro2中触发整个TaskGroup的取消，但TaskGroup没有公开的cancel()方法，以下是两种可行的解决方式：

方法一：利用TaskGroup的自动取消机制 #

asyncio.TaskGroup的核心特性之一是：当任何子任务抛出未捕获异常（包括CancelledError）时，它会自动取消所有剩余子任务，等待它们完成后退出async with代码块。基于这个特性，我们可以让coro2在需要取消时主动取消自身：

import asyncio
import random

async def coro1():
    while True:
        await asyncio.sleep(1)
        print("coro1")

async def coro2():
    while True:
        await asyncio.sleep(1)
        if random.random() < 0.1:
            print("dead")
            # 主动取消当前任务，触发TaskGroup的全局取消逻辑
            await asyncio.current_task().cancel()
        else:
            print("Survived another second")

async def main():
    async with asyncio.TaskGroup() as tg:
        tg.create_task(coro1())
        tg.create_task(coro2())

asyncio.run(main())

方法二：直接取消所有子任务 #

如果需要更直接的控制，可以在创建任务时保存所有子任务的引用，在coro2中遍历这些引用逐个取消：

import asyncio
import random

tasks: list[asyncio.Task] = []

async def coro1():
    while True:
        await asyncio.sleep(1)
        print("coro1")

async def coro2():
    while True:
        await asyncio.sleep(1)
        if random.random() < 0.1:
            print("dead")
            # 遍历并取消所有子任务
            for task in tasks:
                task.cancel()
        else:
            print("Survived another second")

async def main():
    global tasks
    async with asyncio.TaskGroup() as tg:
        tasks = [
            tg.create_task(coro1()),
            tg.create_task(coro2())
        ]
    tasks = []

asyncio.run(main())

说明 #

• 方法一更贴合TaskGroup的设计理念，利用其内置的异常处理逻辑自动完成任务组的清理和退出。
• 方法二更灵活，适合需要对取消过程做额外处理的场景。

内容的提问来源于stack exchange，提问作者Jeffrey Chen