Python asyncio.gather 怎样收集部分任务异常

需要让一批独立任务各自交回结果、同时保留其中的普通异常，可以使用 await asyncio.gather(*aws, return_exceptions=True)。返回列表按传入的 awaitable 顺序排列，成功位置放返回值，失败位置放异常对象；调用方还要逐项识别，不能把“获得了列表”当作“全部成功”。

下面制作一个三条记录的小型汇总脚本，不涉及网络请求、部署或数据库。行为依据 Python 官方协程与任务文档：https://docs.python.org/3/library/asyncio-task.html#asyncio.gather。

把三项任务定义成可独立失败的批次

输入条目为 A、B、C。每个条目由 inspect_item 独立处理，其中 B 故意抛出 ValueError，A、C 返回状态字典。批次允许部分失败，所以 collect_report 需要保留另外两个成功结果，而不是把整批直接判为失败。

任务身份保存在 items 中，原始结果保存在 results 中，通过 zip 按位置配对。结果位置代表输入身份，不代表完成先后；若任务耗时不同，也不能拿结果列表推断谁先结束。这里三个标识互不重复，真实业务如有同名条目，应另带唯一请求标识。

准备一个只依赖标准库的脚本

示例以 Python 3.11 及以上为运行环境，只使用标准库 asyncio，无需安装第三方包。将下节完整代码保存为 batch_report.py。inspect_item 的成功契约固定为字典；这点可以避免“任务正常返回一个异常对象”和“任务真的抛出异常”在结果列表中难以区分。

asyncio.sleep(0) 只是让出事件循环的控制权，不是性能测试，也不模拟真实请求超时。示例不承诺吞掉所有控制类异常，更不应拿 KeyboardInterrupt、SystemExit 来做业务容错测试。

用异常返回模式构建汇总器

import asyncio


async def inspect_item(item):
    # 用本地异步等待模拟任务，不访问任何外部服务。
    await asyncio.sleep(0)
    if item == "B":
        raise ValueError("条目内容无效")
    return {"item": item, "status": "ready"}


async def collect_report(items):
    # 约定每个任务成功时只返回字典，不返回异常对象。
    results = await asyncio.gather(
        *(inspect_item(item) for item in items),
        return_exceptions=True,
    )
    report = {"success": [], "failed": [], "cancelled": []}
    for item, result in zip(items, results):
        # 子任务取消不是普通 Exception，必须单独识别。
        if isinstance(result, asyncio.CancelledError):
            report["cancelled"].append({"item": item})
        elif isinstance(result, Exception):
            report["failed"].append({
                "item": item,
                "error_type": type(result).__name__,
                "message": str(result),
            })
        elif isinstance(result, BaseException):
            # 不把其他控制类异常冒充业务成功。
            raise result
        else:
            report["success"].append(result)
    return report


async def main():
    report = await collect_report(["A", "B", "C"])
    print(report)


if __name__ == "__main__":
    # 只在独立脚本中创建顶层事件循环。
    asyncio.run(main())

gather 接收协程对象时会自动将其调度为任务，普通异常在 return_exceptions=True 下进入 results。分类器依次处理取消、普通异常与成功值；CancelledError 直接继承 BaseException，因此只写 isinstance(result, Exception) 会漏掉子任务取消。

三种报告容器是 success、failed、cancelled。success 保存业务值，failed 保存条目身份、错误类型和消息，cancelled 只登记取消状态。不要把异常对象直接交给 JSON 序列化器；如需完整诊断，可在内部日志保存异常信息，但不应把敏感错误详情原样公开给用户。

按固定输入核对预期报告

在普通终端环境中，读者可以使用下面命令运行；本文未执行示例，以下内容是按给定输入推导的预期，而非实测日志。

# 使用本机 Python 3.11 或更高版本执行独立脚本。
python3 batch_report.py

预期报告如下。A、C 出现在成功列表，B 出现在失败列表，取消列表为空：

{'success': [{'item': 'A', 'status': 'ready'}, {'item': 'C', 'status': 'ready'}], 'failed': [{'item': 'B', 'error_type': 'ValueError', 'message': '条目内容无效'}], 'cancelled': []}

可进一步把输入改为 A、C，此时失败列表应为空；改为只含 B，成功列表应为空。这里的数字仅是示例输入数量，不能用于评价吞吐量。代码没有主动取消任何子任务，所以主示例不应出现取消记录。

嵌入已有异步服务时保留取消语义

已有事件循环的 Web 服务或 Notebook 应在异步上下文里直接 await collect_report(items)，不要在其中再次调用 asyncio.run。返回报告只是这一批任务的汇总，需要由调用层决定部分失败是否可接受，而不是无条件返回业务成功。

某个子 Task 被取消时，gather 不会仅因这一个子任务取消而取消其他成员；启用异常返回模式后，可以在相应结果位置识别 CancelledError。但等待 gather 的调用方被取消、进而使 gather 自身取消时，尚未完成的成员会收到取消请求，return_exceptions=True 也不会把整个操作的取消变成普通报告。collect_report 没有捕获取消异常，正是为了保留这一传播。

取消请求不等于底层工作已经立即停止，实际清理由协程配合完成。任务应使用 finally 释放自身资源；不要在工作协程里随意吞掉 CancelledError。批次也没有内置超时，只要有成员一直不结束，就可能持续等待，生产任务需另行明确单项时限和并发数量。

如果去掉 return_exceptions=True，默认模式会把最先抛出的普通异常传播给 await gather 的调用方，但并不会因此自动取消其他成员。若随后顶层 asyncio.run 因异常退出，其关闭事件循环期间的清理又可能取消仍在运行的任务；不能把这两件事混为一谈。

确认结果边界与常见疑问

为什么拿到结果列表后仍要判断异常？

该参数改变的是异常的交付方式，不会修复失败任务。报告是否满足业务要求，应根据 failed、cancelled 和必需成功项共同判断。

可以用 if result 判断成功吗？

不可以。正常结果可能是 0、空字符串或 None，异常对象也不能靠真值可靠分类。本文先识别异常类型，剩下的值再按业务契约处理。

失败任务会自动重试吗？

不会。若要重试，必须另行定义次数、退避和幂等性；不能把同一个已经等待过的协程对象重新当作新任务提交。本文只解决原批次的结果收集，不加入重试逻辑。

对于允许部分成功的小批独立任务，关键写法是启用异常返回模式、保持输入身份对应、逐项分类，同时让调用方取消继续传播。这样得到的是一份明确的批次报告，而不是隐藏失败的成功列表。