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

还可以补一个常见边界：给整个任务组设置总时限。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 等待清理完成的保证。

还有一个与 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 语义。

把这类行为写成回归测试时，不建议断言打印顺序：调度时序可能变化。更稳定的做法是直接检查兄弟任务的取消状态、清理信号和异常组内容。

环境：Python 3.11+；仅使用标准库。创建 test_task_group.py：

import asyncio
import unittest


class TaskGroupTest(unittest.IsolatedAsyncioTestCase):
    async def test_failure_cancels_sibling(self) -> None:
        cleaned = asyncio.Event()

        async def slow() -> None:
            try:
                await asyncio.sleep(60)
            finally:
                cleaned.set()

        async def broken() -> None:
            await asyncio.sleep(0)
            raise ValueError("boom")

        with self.assertRaises(ExceptionGroup) as caught:
            async with asyncio.TaskGroup() as group:
                slow_task = group.create_task(slow())
                group.create_task(broken())

        self.assertTrue(slow_task.cancelled())
        self.assertTrue(cleaned.is_set())
        self.assertTrue(
            any(
                isinstance(error, ValueError)
                and str(error) == "boom"
                for error in caught.exception.exceptions
            )
        )


if __name__ == "__main__":
    unittest.main()

运行 python -m unittest -v test_task_group.py。IsolatedAsyncioTestCase 会为测试管理独立事件循环；三个断言分别覆盖 fail-fast 取消、finally 清理以及原始异常没有丢失。

这里还有一个容易误判的时间语义：asyncio.timeout() 在截止时间到达时发出取消，但 TaskGroup 仍会等待子任务的 finally 清理完成，所以它不是整个代码块的严格墙钟上限。

环境：Python 3.11+；Linux、macOS 或 Windows；仅使用标准库。下面脚本可直接运行：

import asyncio
import time


async def slow() -> None:
    try:
        await asyncio.sleep(60)
    finally:
        await asyncio.sleep(0.2)


async def main() -> None:
    started = time.monotonic()
    try:
        async with asyncio.timeout(0.1):
            async with asyncio.TaskGroup() as group:
                group.create_task(slow())
    except TimeoutError:
        pass

    print(f"elapsed: {time.monotonic() - started:.1f}s")


asyncio.run(main())

运行 python timeout_cleanup.py，输出通常接近 elapsed: 0.3s，而不是 0.1s：前 0.1s 到期后，任务组还等待了约 0.2s 的异步清理。实际使用时应让清理路径本身有界，并把这里的 timeout 理解为“开始取消的期限”；结构化并发仍优先保证子任务完成收尾。