如何使用Python 3.11的异常组功能处理自动化任务中的多点故障？-Python教程-PHP中文网

exceptiongroup 是 python 3.11 引入的用于结构化组织并发或批量操作中独立发生的多个异常的容器，except* 是专用于匹配其内部子异常的语法，支持递归匹配、尽力而为，并仅对 exceptiongroup 及其子类生效。

如何使用python 3.11的异常组功能处理自动化任务中的多点故障？

什么是 `ExceptionGroup` 和 `except*`？

Python 3.11 引入的 ExceptionGroup 不是“多个异常拼一起”，而是把并发或批量操作中**独立发生的多个异常**组织成一个结构化容器；except* 是专门用来匹配这种嵌套异常结构的语法，它不追求“捕获一个就退出”，而是尝试匹配所有符合条件的子异常。

常见误用是把它当 try/except 的增强版来兜底——结果发现没进 except* 分支，因为没触发“匹配多个”的条件。

ExceptionGroup 必须显式构造（比如 ExceptionGroup("task failed", [exc1, exc2])），不会自动产生
except* 只对 ExceptionGroup 或其子类生效，对普通异常无效
匹配是“尽力而为”：只要子异常里有至少一个满足类型条件，该 except* 就会执行，并只拿到那些匹配的子异常

在 `asyncio.gather` 中触发 `ExceptionGroup`

asyncio.gather 在 Python 3.11+ 中默认启用 return_exceptions=False，一旦任一协程抛异常，整个调用就以 ExceptionGroup 形式失败——这是最典型的自动化任务多点故障场景。

import asyncio
<p>async def fetch(url):
if "fail" in url:
raise ValueError(f"Failed to fetch {url}")
return f"OK: {url}"</p><p>async def main():
try:
results = await asyncio.gather(
fetch("<a href="https://www.php.cn/link/e75a2d435455f0626ccf0af67216e76f">https://www.php.cn/link/e75a2d435455f0626ccf0af67216e76f</a>"),
fetch("<a href="https://www.php.cn/link/a4f32d3183be54c56a76dbfc8b992eb3">https://www.php.cn/link/a4f32d3183be54c56a76dbfc8b992eb3</a>"),
fetch("<a href="https://www.php.cn/link/6fe0fa22eca4dea0f85b883b75c76a34">https://www.php.cn/link/6fe0fa22eca4dea0f85b883b75c76a34</a>"),
fetch("<a href="https://www.php.cn/link/91906afbe4627f854c07693c7f4264cb">https://www.php.cn/link/91906afbe4627f854c07693c7f4264cb</a>"),
)
except* ValueError as eg:
print(f"Caught {len(eg.exceptions)} value errors")
for e in eg.exceptions:
print(f"  → {e}")
</p>

注意：except* 不会中断其他协程——gather 已完成的部分结果不会丢失，但未完成的会被取消。如果想保留全部结果（包括异常），得用 return_exceptions=True，但那样你就得自己遍历结果并识别 Exception 实例，不再走 ExceptionGroup 流程。

手动构造 `ExceptionGroup` 处理批处理失败

不是所有多点故障都来自协程。比如你用 ThreadPoolExecutor 并行处理一批文件，每个线程可能独立报错——这时需要手动收集异常、构造 ExceptionGroup。

提示词大师-python版

图片提示词生成器？不止如此。马甲系统 —— 把脑海中的画面，翻译成AI能理解的专业表达。用得越多，它越懂你：首次需要多问几句确认方向，用久了几乎一说就懂。用得越多，它越快：缓存机制让后续对话越来越省。 RAG进化：成功案例持续入库，越跑越聪明。输入「新手指南」查看完整功能介绍

下载

别直接用 raise ExceptionGroup(...) 后再 except*，那只是测试；真实场景要先 try/except 捕获每个子任务异常，存进列表
构造时传入的异常列表不能为空，否则会报 ValueError: exceptions must not be empty
建议给 ExceptionGroup 的 message 参数起有意义的名字，比如 "file processing failed"，方便日志归类

示例片段：

from concurrent.futures import ThreadPoolExecutor
from exceptiongroup import ExceptionGroup  # Python def process_file(path):
if path.endswith(".bad"):
raise OSError(f"Permission denied: {path}")
return f"Processed {path}"<p>errors = []
results = []
with ThreadPoolExecutor() as ex:
futures = [ex.submit(process_file, p) for p in ["a.txt", "b.bad", "c.txt", "d.bad"]]
for f in futures:
try:
results.append(f.result())
except BaseException as e:
errors.append(e)</p><p>if errors:
raise ExceptionGroup("batch processing failed", errors)
</p>

`except*` 嵌套和性能影响

except* 支持嵌套：一个 ExceptionGroup 里可能包含另一个 ExceptionGroup（比如某子任务本身也用了 gather）。这时 except* ValueError 会递归进入所有层级找匹配项。

但要注意两点：

它不等价于 except (ValueError, TypeError)：后者是“或”关系，前者是“在异常树里找所有匹配的叶子节点”
深度嵌套 + 大量异常时，except* 的匹配开销略高于普通 except，不过对几十个异常基本无感；真正卡住的是你后续怎么处理——比如逐个打印堆栈，会显著拖慢日志输出
别在 except* 块里再抛出新异常，除非你明确想覆盖原 ExceptionGroup；否则容易丢失原始上下文

复杂点在于：异常组里的每个子异常仍保留完整 traceback，但默认打印时只显示第一层。要用 eg.print_traceback() 才能看到全部细节——这点很容易被忽略，导致调试时以为“只失败了一个”。