Python TaskGroup 怎么主动终止整组任务

Python 的 asyncio.TaskGroup 没有原生 terminate() 方法。要主动结束整组任务，官方给出的做法是在仍处于活动状态的 TaskGroup 中加入一个“终止任务”，让它抛出专用异常；TaskGroup 随即取消其余成员、等待清理完成，再把异常放进 ExceptionGroup。调用方最后用 except* 只抑制这类终止异常。

Python 官方文档：https://docs.python.org/3.14/library/asyncio-task.html

TaskGroup 的主动终止模型

TaskGroup 的关键规则是：组内任一任务首次抛出非 CancelledError 异常时，其余未完成任务会被取消；退出上下文前，TaskGroup 会等待它们结束，再把非取消异常组合后抛出。专用终止异常正是借用这条规则，但它只表达“按计划结束整组”，不应与业务失败共用同一种异常。

最小实现：注入一个专用异常任务

下面的例子启动两个持续工作的协程，等待约 1.2 秒后，把终止协程加入同一个 TaskGroup。终止协程一运行就抛出 TerminateTaskGroup，两个工作任务在后续取消点收到取消请求，并执行各自的 finally。

import asyncio


class TerminateTaskGroup(Exception):
    """只用于请求结束当前 TaskGroup。"""


async def force_terminate_task_group() -> None:
    # 专用异常会触发 TaskGroup 的兄弟任务取消机制。
    raise TerminateTaskGroup()


async def worker(name: str) -> None:
    try:
        while True:
            print(f"{name}: working")
            await asyncio.sleep(0.5)  # await 是可响应取消的位置。
    finally:
        # 在这里关闭连接、释放锁或刷新缓冲区，不吞掉取消异常。
        print(f"{name}: cleanup")


async def main() -> None:
    try:
        async with asyncio.TaskGroup() as group:
            group.create_task(worker("A"))
            group.create_task(worker("B"))

            await asyncio.sleep(1.2)
            # TaskGroup 仍活动时，加入抛出专用异常的终止任务。
            group.create_task(force_terminate_task_group())
    except* TerminateTaskGroup:
        # 只抑制主动终止信号，其他异常分组仍会向外传播。
        pass


asyncio.run(main())

这段代码的判断标准不是“两个 worker 立刻消失”，而是：终止任务抛出异常后，两个 worker 收到取消请求；TaskGroup 等待它们执行清理；except* 最后只消费 TerminateTaskGroup。asyncio 的取消是协作式的，任务通常在下一次 await 时才观察到取消。

由外部条件触发整组终止

真实服务通常由事件、健康检查或业务状态触发停止。可以在组内增加一个监督任务，它等待 asyncio.Event，条件成立后再向仍活动的 TaskGroup 注入终止任务。

import asyncio


class TerminateTaskGroup(Exception):
    """表示一次预期中的整组终止请求。"""


async def force_terminate_task_group() -> None:
    # 抛出非 CancelledError，触发同组任务取消。
    raise TerminateTaskGroup()


async def stop_when_requested(
    group: asyncio.TaskGroup,
    stop_event: asyncio.Event,
) -> None:
    # 监督任务与业务任务处于同一个结构化并发边界内。
    await stop_event.wait()
    group.create_task(force_terminate_task_group())


async def service(name: str) -> None:
    try:
        while True:
            await asyncio.sleep(0.5)  # 周期性让出控制权以响应取消。
            print(f"{name}: heartbeat")
    finally:
        # 清理完成后让 CancelledError 自然继续传播。
        print(f"{name}: closed")


async def main() -> None:
    stop_event = asyncio.Event()

    async def request_stop() -> None:
        # 这里模拟外部条件到达，生产环境可替换为信号或状态监听。
        await asyncio.sleep(2)
        stop_event.set()

    try:
        async with asyncio.TaskGroup() as group:
            group.create_task(service("api"))
            group.create_task(service("consumer"))
            group.create_task(stop_when_requested(group, stop_event))
            group.create_task(request_stop())
    except* TerminateTaskGroup:
        # 预期终止已完成；其他业务异常不会被这一分支吞掉。
        pass


asyncio.run(main())

关键限制是 group.create_task() 必须发生在 TaskGroup 仍活动时。组尚未进入、已经退出或正在关闭时再添加协程都会失败，所以不要把 TaskGroup 对象长期保存到结构化并发边界之外。

取消边界怎么选

专用异常任务只适用于“结束整个组”。如果只想停止一个成员，应保存 create_task() 返回的 Task 并调用该任务的 cancel()；如果是上层调用方取消当前协程，应完成清理后继续传播 CancelledError，不要把外部取消伪装成一次正常结束。

三个容易踩中的坑

1. 用 CancelledError 充当终止信号

不要让终止任务主动抛出 CancelledError。TaskGroup 会把取消异常作为特殊情况处理，它不等同于“某个成员失败并取消兄弟任务”。专用的普通异常才能清楚表达主动终止意图。

2. 在 worker 中吞掉取消异常

协程可以用 try/finally 做清理；如果确实捕获了 CancelledError，通常也应在清理后重新抛出。TaskGroup 和 asyncio.timeout() 都依赖取消实现结构化并发，静默吞掉取消可能让它们表现异常。

3. 误以为取消能中断同步阻塞

取消请求不会抢占正在运行的普通 Python 代码。长循环需要定期 await，阻塞 I/O 可以考虑 asyncio.to_thread()，CPU 密集任务则应交给进程池或拆成能让出控制权的单元。

运行检查

检查实现时，不要只看主协程是否返回。应确认两个工作任务都执行了清理、没有“Task exception was never retrieved”日志、预期终止不会被记录成业务故障，并且真实 worker 异常仍能越过 except* TerminateTaskGroup。还可以在测试中让一个 worker 同时抛出真实异常，验证 except* 只移除专用终止异常的子组。

相关问题

TaskGroup 为什么不直接提供 terminate 方法？

标准库目前没有原生终止接口。官方文档明确给出了“添加抛异常任务并忽略该专用异常”的实现方式。

except* 和普通 except 有什么区别？

TaskGroup 可能同时收集多个异常并抛出 ExceptionGroup。except* 能按异常类型拆分处理，只消费匹配的 TerminateTaskGroup，其余异常继续传播。

终止后 finally 一定会执行吗？

任务在可响应取消的位置收到 CancelledError 后会展开栈，正常进入 finally。但若任务一直同步阻塞、强制结束进程或清理代码自身卡住，就不能期待它及时完成。

可以从 TaskGroup 外部直接调用 group.create_task 吗？

只有 TaskGroup 仍处于活动状态时才可以。更稳妥的做法是把外部信号转成 Event，由组内监督任务负责注入终止任务。