Python asyncio TaskGroup 如何收口异常：取消传播与部分结果处理

批量请求三个下游服务时，最麻烦的情况不是某一个请求失败，而是失败之后另外两个任务还在继续写入共享状态。Python 3.11 引入的 asyncio.TaskGroup 会把这件事收紧：一个子任务抛出未处理异常，其他兄弟任务会被取消，离开代码块时再以聚合异常的形式交给调用方。

先看一个会失控的批量调用场景

假设一次批处理要同时读取库存、价格和优惠券。库存任务先成功，价格任务抛出异常，优惠券任务却还在等待网络返回。如果外层只捕获价格异常，最终很容易留下“库存已写入、价格未更新、优惠券稍后又回来覆盖状态”的半成品。

TaskGroup 的关键不是让请求更快，而是把这组任务的生命周期绑定在同一个上下文里。代码块退出前，组内任务必须结束；其中一个任务失败时，未完成的兄弟任务会收到取消信号。

TaskGroup 的最小可用写法

import asyncio

async def fetch_stock():
    await asyncio.sleep(0.05)
    return {"sku": "A-100", "stock": 8}

async def fetch_price():
    await asyncio.sleep(0.02)
    raise RuntimeError("price service unavailable")

async def fetch_coupon():
    try:
        await asyncio.sleep(0.30)
        return {"discount": 10}
    finally:
        print("coupon task cleanup")

async def load_product():
    async with asyncio.TaskGroup() as group:
        stock_task = group.create_task(fetch_stock())
        price_task = group.create_task(fetch_price())
        coupon_task = group.create_task(fetch_coupon())

    return stock_task.result(), price_task.result(), coupon_task.result()

async def main():
    try:
        result = await load_product()
        print(result)
    except* Exception as errors:
        print("batch failed:", errors)

asyncio.run(main())

运行后，fetch_price() 的异常会让 fetch_coupon() 被取消，控制台仍会执行它的清理分支。load_product() 不会在组内任务尚未收口时提前返回，这就是它和“手动 create_task 后随便 gather”最重要的边界。

取消传播：为什么 finally 比 except 更可靠

兄弟任务收到的是 CancelledError。如果任务持有连接、临时文件或批次锁，清理动作应放在 finally 中；如果确实需要记录取消原因，可以先捕获，再继续抛出：

async def worker(resource):
    try:
        await resource.run()
    except asyncio.CancelledError:
        await resource.mark_cancelled()
        raise
    finally:
        await resource.close()

这里的 raise 不能省略。吞掉取消异常会让 TaskGroup 误以为任务正常结束，调用方就可能把一个已经取消的批次当成成功。只有在非常明确的恢复场景下，才考虑 uncancel() 等更特殊的处理。

想保留部分结果，先区分“结果”与“成功状态”

TaskGroup 失败并不意味着所有已经完成的计算都不存在。可以让每个任务把自己的结果写入一个普通字典，但这个字典只能作为诊断或重试输入，不能直接代表批次成功。

async def run_batch():
    results = {}

    async def capture(name, operation):
        try:
            results[name] = await operation()
        except asyncio.CancelledError:
            results[name] = {"state": "cancelled"}
            raise
        except Exception as exc:
            results[name] = {"state": "failed", "error": str(exc)}
            raise

    try:
        async with asyncio.TaskGroup() as group:
            group.create_task(capture("stock", fetch_stock()))
            group.create_task(capture("price", fetch_price()))
            group.create_task(capture("coupon", fetch_coupon()))
    except* Exception as errors:
        return {"state": "failed", "results": results, "errors": str(errors)}

    return {"state": "ok", "results": results}

这个写法有两个收口点：每个任务负责留下可解释的局部状态，外层负责决定整批是 ok 还是 failed。不要因为字典里已经有库存结果，就把整个订单批次写成完成。

异常聚合与重试边界

TaskGroup 离开上下文时可能抛出 ExceptionGroup。使用 except* 可以按异常类型分拣，例如把网络错误放进一次批次重试，把数据校验错误直接送入人工处理。

try:
    await run_batch()
except* TimeoutError as timeout_errors:
    await schedule_retry("network", timeout_errors)
except* ValueError as data_errors:
    await record_bad_input(data_errors)

重试也要有边界：只重试可恢复的下游错误，并携带批次 ID 做幂等；不要在 except* 中无条件再次创建同一组任务，否则一次故障会被放大成重试风暴。

三个常见误区

把 TaskGroup 当成 gather 的换名字

gather() 的异常与取消语义要结合参数理解，而 TaskGroup 更强调结构化并发的生命周期。选哪个取决于调用方是否需要“一组任务绑定在一起”的失败边界。

在取消分支里只打印日志

日志不能替代状态清理。连接、事务和临时资源需要明确关闭；处理完后继续抛出取消异常，才能让外层正确收口。

把部分结果直接当成功结果返回

部分结果适合诊断、补偿和重试。对外返回前应有一个明确的批次状态字段，否则调用方很难区分“只完成了一半”和“全部成功”。

相关问题

TaskGroup 从哪个 Python 版本开始可用？

它是 Python 3.11 标准库 asyncio 的能力。更老的运行时需要升级，或继续使用现有并发方案并自行约定取消和收口规则。

一个任务失败时，已经完成的任务会被撤销吗？

TaskGroup 不会自动撤销外部副作用。它只负责任务生命周期和异常传播；数据库写入、消息发送等操作仍要用事务、幂等键或补偿逻辑保护。

什么时候更适合使用 gather？

如果任务之间相互独立，调用方希望按返回顺序收集结果，且已有清晰的异常策略，gather() 仍然直接。需要绑定取消、清理和整体完成边界时，再优先考虑 TaskGroup。

把收口规则写进批处理协议

TaskGroup 真正解决的是边界：谁创建任务，谁等待任务，谁决定批次状态，谁负责清理。把这四件事固定下来，再把部分结果和失败原因作为独立字段保存，并发代码就不容易在异常路径上留下“看起来完成、实际上未完成”的状态。