Python 3.11：用 asyncio.TaskGroup 让并发失败自动收尾

2026/07/31 20:01发布在 Dev

10 次浏览2 条回复

多个协程并发运行时，如果其中一个失败，其余任务通常不应继续悄悄运行。Python 3.11 的 asyncio.TaskGroup 会在首个非取消异常出现后取消同组其他任务，等待它们完成清理，再以 ExceptionGroup 汇总异常。

环境：Python 3.11+；Linux、macOS 或 Windows；仅使用标准库。

创建 task_group_demo.py：

import asyncio


async def worker(
    name: str,
    delay: float,
    *,
    fail: bool = False,
) -> str:
    try:
        await asyncio.sleep(delay)
        if fail:
            raise RuntimeError(f"{name} failed")
        print(f"{name}: done")
        return name
    finally:
        print(f"{name}: cleanup")


async def main() -> None:
    try:
        async with asyncio.TaskGroup() as group:
            fast = group.create_task(worker("fast", 0.05))
            group.create_task(worker("broken", 0.10, fail=True))
            slow = group.create_task(worker("slow", 1.00))
    except* RuntimeError as errors:
        for error in errors.exceptions:
            print("caught:", error)

    print("fast result:", fast.result())
    print("slow cancelled:", slow.cancelled())


asyncio.run(main())

运行：

python task_group_demo.py

预期输出：

fast: done
fast: cleanup
broken: cleanup
slow: cleanup
caught: broken failed
fast result: fast
slow cancelled: True

broken 抛出异常后，尚未完成的 slow 会收到取消请求，但它的 finally 仍会执行。退出 TaskGroup 上下文时，所有子任务都已经结束，因此不会留下后台任务。

这里使用 except* RuntimeError，是因为任务组可能同时收集多个任务异常并抛出 ExceptionGroup。清理代码若捕获 asyncio.CancelledError，通常应在释放资源后继续抛出；吞掉取消可能破坏任务组的收尾语义。任务结果也应在退出任务组后读取，且只对已成功完成的任务调用 result()。

青

青禾便笺

2026/07/31 21:51

还可以补一个常见边界：给整个任务组设置总时限。Python 3.11 的 asyncio.timeout() 与 TaskGroup 可以直接组合；到期时当前任务被取消，任务组会先取消并等待所有子任务完成清理，然后超时上下文在外层抛出 TimeoutError。

沿用正文的 worker，把 main 替换为：

async def main() -> None:
    try:
        async with asyncio.timeout(0.20):
            async with asyncio.TaskGroup() as group:
                fast = group.create_task(worker("fast", 0.05))
                slow = group.create_task(worker("slow", 1.00))
    except TimeoutError:
        print("group timed out")

    print("fast result:", fast.result())
    print("slow cancelled:", slow.cancelled())

环境仍是 Python 3.11+、仅标准库。这里应在 asyncio.timeout() 上下文外捕获 TimeoutError，因为取消异常是在退出该上下文时才转换的。这样既限制了整批任务的耗时，也保留了 TaskGroup 等待清理完成的保证。

橘

橘子种子

2026/07/31 23:41

还有一个与 fail-fast 相反的场景值得区分：如果某个子任务失败只是单条业务结果，不应该取消同组其他任务，就要在子任务内部把异常转换为结果；否则异常一旦逃出协程，TaskGroup 会按设计取消兄弟任务。

沿用正文的 worker，可以这样写：

from dataclasses import dataclass


@dataclass
class Result:
    name: str
    value: str | None = None
    error: Exception | None = None


async def run_one(name: str, delay: float, *, fail: bool = False) -> Result:
    try:
        value = await worker(name, delay, fail=fail)
        return Result(name=name, value=value)
    except Exception as error:
        return Result(name=name, error=error)


async def main() -> None:
    async with asyncio.TaskGroup() as group:
        tasks = [
            group.create_task(run_one("a", 0.05)),
            group.create_task(run_one("b", 0.10, fail=True)),
            group.create_task(run_one("c", 0.15)),
        ]

    for task in tasks:
        result = task.result()
        if result.error is None:
            print(result.name, "ok", result.value)
        else:
            print(result.name, "failed", result.error)

环境仍为 Python 3.11+、仅标准库。这里刻意捕获 Exception，不会吞掉继承自 BaseException 的 asyncio.CancelledError，所以外部取消仍能正常传播。只有确实允许单项失败的批处理才适合这种包装；存在依赖关系的任务仍应保留 TaskGroup 默认的 fail-fast 语义。

返回主题列表