当前位置：首页 > 文章列表 > 文章 > python教程 > Python asyncio TaskGroup 如何收口异常：取消传播与部分结果处理

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

来源：17golang原创 2026-08-27 07:22:42 0浏览收藏

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

要点速览

TaskGroup 适合“一组任务要么一起完成、要么整体收口”的并发批次。
兄弟任务收到取消后，清理逻辑应放在 finally，不要吞掉 CancelledError。
需要保留成功结果时，把结果写入批次上下文，并在外层用 except* Exception 统一整理。
失败任务已经决定批次结果后，不要继续把半成品标记为可发布或可结算。

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

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

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”最重要的边界。

Python TaskGroup 中价格任务失败后取消优惠券任务并进入清理阶段的数据生命周期示意图

取消传播：为什么 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。不要因为字典里已经有库存结果，就把整个订单批次写成完成。

Python TaskGroup 将已完成、失败和取消结果分开记录并由外层统一收口的示意图

异常聚合与重试边界

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* 中无条件再次创建同一组任务，否则一次故障会被放大成重试风暴。