Python asyncio.gather 异常为什么会提前结束：return_exceptions 与任务取消边界

批量调用三个异步接口时，最容易踩的坑就是：其中一个任务抛出异常，await asyncio.gather(...) 立刻向外抛出报错，但另外两个没跑完的任务日志还在后台继续输出。gather 默认只会把第一个触发的异常透传给调用方，根本不会自动帮你终止剩下的任务；如果你的业务要求“一个环节失败就立刻停掉整组并发逻辑”，必须手动把协程存成 Task 对象、逐个发取消信号，再等待所有任务执行完清理逻辑才算完成。

先复现一个“外层失败、内层未停”的现场

我们先写三个有明确时间差的测试任务：fast_fail 执行0.2秒后主动抛出失败，两个慢任务每隔0.3秒打印一次运行进度，这个时间差足够把异步并发里的隐藏问题完全暴露出来。

import asyncio

async def fast_fail():
    await asyncio.sleep(0.2)
    raise RuntimeError("inventory service unavailable")

async def slow_job(name):
    try:
        for step in range(4):
            await asyncio.sleep(0.3)
            print(name, "step", step + 1)
        return name + " done"
    finally:
        print(name, "cleanup")

async def main():
    tasks = [fast_fail(), slow_job("price"), slow_job("stock")]
    try:
        await asyncio.gather(*tasks)
    except RuntimeError as exc:
        print("caller got:", exc)
    await asyncio.sleep(0.8)

asyncio.run(main())

运行这段代码你会先看到 caller got: inventory service unavailable 抛出的报错，之后还可能陆续看到 price 和 stock 打印的进度日志。外层协程已经跳转到异常处理分支，不代表 gather 调度创建的其他工作流会被统一撤销。不用急着给所有异常随便套一层 try 掩盖，先想清楚你当前的业务场景到底要“收集所有任务结果”还是“一个失败就立刻终止全组”。

两种策略的边界：收集异常，还是取消整组任务

参数 return_exceptions 改变的只有返回结果的包装形态，完全没有修改任务本身的生命周期规则。把两种常用处理逻辑放在一张小表里对照，能最大程度避免用错场景。

如果每个并发任务的业务逻辑完全独立，第一种写法用起来会更省心；但如果后续任务继续跑会产生重复扣款、冗余无效写入或者超出预算的高额请求，绝对不能把 return_exceptions=True 当成什么“安全防护开关”，它仅仅是把异常挪到结果列表里返回而已。

实现失败即止：取消、等待和保留原异常

下面的示例代码会先把协程手动包装成 Task 存起来。任意一个任务抛出异常后，循环调用每个 Task 实例的 cancel() 方法，再通过 gather(..., return_exceptions=True) 等待所有 Task 真正执行完毕。这第二次等待非常关键，它给了每个被取消的任务机会，在 finally 里正常关闭连接、释放信号量或者删除已经生成的临时文件。

async def run_batch_fail_fast():
    tasks = [
        asyncio.create_task(fast_fail(), name="fast_fail"),
        asyncio.create_task(slow_job("price"), name="price"),
        asyncio.create_task(slow_job("stock"), name="stock"),
    ]
    try:
        return await asyncio.gather(*tasks)
    except BaseException:
        for task in tasks:
            if not task.done():
                task.cancel()
        await asyncio.gather(*tasks, return_exceptions=True)
        raise

async def main():
    try:
        await run_batch_fail_fast()
    except RuntimeError as exc:
        print("batch failed:", exc)

这里捕获 BaseException 是为了让调用方主动触发整批次取消时也能正常进入清理分支；真正的原始业务异常仍旧会在最后通过 raise 原样向外抛出。不要在清理阶段直接再次等待可能抛出错误的 Task 实例，否则第一个触发的业务异常很可能被后续清理阶段抛出的其他异常覆盖。

把超时和外层取消纳入同一条验收路径

线上生产环境跑的代码，通常还会在外层套一层全局超时。超时本身就是取消信号的一种常见表现形式，验收逻辑不能只测“某个接口返回业务错误”这一种场景，还要验证超时触发后，连接池占用、临时文件、限流传入的令牌这些资源是不是都能正常回收。

async def guarded_batch():
    try:
        async with asyncio.timeout(1.0):
            return await run_batch_fail_fast()
    except TimeoutError:
        print("batch timeout")
        raise

• 所有任务正常完成：三个任务的结果都正常返回，每个任务的清理日志各打印一次。
• 子任务运行失败：调用方收到原始的 RuntimeError，所有还没跑完的 Task 都进入已取消状态。
• 外层触发超时：调用方收到 TimeoutError，子任务注册的 finally 仍旧可以正常执行完成。
• 上游逻辑主动发起取消：不要私自吞掉 CancelledError，否则上层传递的停止信号会被误判为业务执行成功。

常见问题

gather 会自动取消其他任务吗？

默认情况下不会因为任意一个子任务抛出普通业务异常就自动终止其余所有任务。需要实现失败即停逻辑的时候，手动保存所有 Task 实例再逐个显式调用 cancel() 才是可靠的写法。

return_exceptions=True 会吞掉异常吗？

它只是把异常对象当作普通元素塞进返回列表，要不要打印日志、要不要做重试、要不要重新向外抛出，仍旧由调用方的逻辑自行决定。全量批处理汇总场景可以放心使用，涉及关键数据写入的流程不要无条件开启。

为什么给任务发了取消信号之后，还要再调用一次 gather？

取消操作本身只是给任务打了一个取消标记发了个通知，任务必须运行到下一个可中断的挂起点才能响应信号，跑完预定义的清理逻辑。再次等待才能确认整组任务已经完全收尾，没有遗留的后台逻辑继续跑。

能不能只保存原始协程对象，不手动创建 Task？

直接把协程丢给 gather 自动调度当然可以，但后续任务出错之后你很难逐个定位检查、单独发取消信号。需要精细管控任务生命周期的场景，直接显式创建 Task 并给它命名，后续排查线上问题的时候会顺畅很多。

验收清单

这类异步并发逻辑的完成标准从来不是“异常能正常打印出来”，而是失败触发、取消响应、资源清理三个状态都可以被观测、可预期。主动做一次故障注入测试，确认没有后台遗留任务继续访问已经关闭的资源，这套逻辑才算真正写到位。