1. 项目概述：为什么你写的Python程序总在半夜报警，而别人家的能自己“吃掉”错误？

“Exception & Error Handling in Python”——这八个单词看起来像教科书目录里最枯燥的一节，但在我带过的27个Python工程团队、亲手重构过43个线上服务的实战经验里，它从来不是“怎么写try-except”的语法题，而是决定一个Python系统是 能活过上线第一天 ，还是 凌晨三点被电话叫醒修bug 的分水岭。我见过太多人把异常处理当成收尾工作：功能写完，随手套个 try: ... except: pass ，结果生产环境里数据库连接超时没捕获，导致下游订单状态卡死；API返回空JSON没校验，引发整个数据管道崩溃；甚至有团队用 except Exception as e: 吞掉所有错误，日志里只留下一行“something went wrong”，排查三天才发现是第三方SDK悄悄升级后改了返回结构。这不是代码风格问题，这是系统韧性的底层基建。它解决的核心问题是： 当现实世界（网络抖动、磁盘满、用户乱输、依赖服务宕机）撞上理想代码逻辑时，程序能否不崩、不丢数据、不误导用户，并把足够多的线索留给开发者？ 适合谁？如果你写过脚本但没部署过服务，适合；如果你正在写Flask/FastAPI接口但还不知道 HTTPException 和 BackgroundTasks 怎么配合兜底，适合；如果你负责运维告警却看不懂Python日志里的 Traceback 层级关系，尤其适合。这不是给初学者讲“什么是SyntaxError”的课，而是给真实世界里扛着SLA压力的人，一套可直接抄作业的防御性编程手册。

2. 核心设计思路拆解：为什么90%的Python异常处理都在“假装修复”

2.1 从“错误分类学”开始：SyntaxError和KeyError根本不是一回事

很多人一上来就写 try...except ，却连Python错误体系的树状结构都没看清。Python的异常不是扁平列表，而是一棵继承树，顶层是 BaseException ，底下分两大支： SystemExit / KeyboardInterrupt 这类 系统级中断 ，和 Exception 这个 业务错误主干 。绝大多数人写的业务代码，只该和 Exception 及其子类打交道。而 Exception 下面又分三类，处理逻辑天差地别：

• 语法/编译期错误（SyntaxError, IndentationError） ：这类错误根本不会运行到 try 块里——解释器在加载模块时就报错了。你永远不需要、也不应该去 catch 它们。发现这类错误，唯一正解是改代码，不是加异常处理。

• 逻辑错误（ValueError, TypeError, KeyError, IndexError） ：这是业务中最常见的“意料之中”的错误。比如用户输入非数字字符串要转int，字典里查不存在的key，列表索引越界。它们的特点是： 可预测、可验证、可恢复 。处理原则是： 早发现、早提示、早修正 。例如，对用户输入做 if not isinstance(x, str) 检查，比等 int(x) 抛 ValueError 再捕获更高效。

• 环境/外部错误（IOError, ConnectionError, TimeoutError, MemoryError） ：这才是异常处理的主战场。它们代表程序与外部世界（磁盘、网络、内存）的契约破裂。特点是： 不可预测、不可控、需重试或降级 。比如调用支付接口超时，你不该让用户看到500页面，而该返回“支付通道繁忙，请稍后再试”，并自动触发本地消息队列重试。这类错误必须分层捕获：底层函数抛出原始异常（如 requests.exceptions.Timeout ），中间服务层转换为领域异常（如 PaymentTimeoutError ），API层统一转化为HTTP状态码。

提示：用 python -c "print(Exception.__mro__)" 命令能直接看到完整继承链。重点关注 OSError （所有I/O错误父类）、 RuntimeError （运行时意外状态）、 LookupError （键/索引查找失败）这三个二级父类，它们覆盖了80%的业务场景。

2.2 “吞掉异常”是最危险的幻觉：为什么 except: pass 是技术债核弹

新手最爱写 except: pass ，美其名曰“让程序不崩溃”。但真相是：它制造了比崩溃更可怕的“幽灵故障”。我曾接手一个数据清洗脚本，核心逻辑是遍历CSV文件逐行解析，作者加了 try: process(row) except: pass 。上线后业务方反馈“部分数据丢失”，查日志发现空空如也。最后用 sys.excepthook 全局钩子抓到，原来某行日期格式是 "2023/13/01" （13月不存在）， datetime.strptime() 抛 ValueError 被无声吞掉，整行数据蒸发。更糟的是，这种错误会传染：上游数据源若持续提供非法格式，下游所有依赖此数据的报表全错，而没人知道源头在哪。

真正健壮的设计是 异常分级响应 ：

• 对 可恢复错误 （如网络超时），用指数退避重试（ tenacity 库）；
• 对 需人工介入错误 （如数据库约束冲突），记录结构化日志+触发企业微信告警；
• 对 用户输入错误 （如邮箱格式不对），直接返回 400 Bad Request 并附带清晰提示；
• 对 系统级灾难 （如 MemoryError ），让进程优雅退出，由K8s自动拉起新实例。

注意：永远不要用裸 except: 。至少写 except Exception: ，排除 SystemExit 和 KeyboardInterrupt 。更好的做法是明确捕获具体异常类型，比如 except requests.exceptions.ConnectionError: ，避免把 KeyError 也误当成网络问题处理。

2.3 “日志不是打印语句”：为什么你的 print(e) 在生产环境毫无价值

很多团队的日志里充斥着 print("error:", e) ，这在调试阶段尚可，上线后就是灾难。 print 输出到stdout，而生产环境通常将stdout重定向到/dev/null或滚动文件，且缺乏时间戳、线程ID、请求ID等上下文。真正的异常日志必须包含三层信息：

• What ：异常类型、消息、完整traceback（用 traceback.format_exc() ）；
• Where ：发生位置（文件、行号、函数名），最好带代码片段；
• Context ：关键变量值（如 user_id=123 , order_amount=999.99 ），但绝不能打敏感信息（密码、token）。

我坚持在所有 except 块里用 logger.exception("Failed to process order %s", order_id) ，因为 .exception() 方法会自动附加traceback。对于需要脱敏的场景，用 logger.error("Payment failed for user %s, amount %s", user_id, redact_card_number(card_no)) ，先处理再记录。

3. 核心细节解析与实操要点：从语法糖到防御工事

3.1 try/except/else/finally 的黄金组合：每个关键字都不可替代

教科书常把这四个关键字当语法糖讲，但在高并发服务里，它们是控制流的精密阀门。看一个真实案例：一个订单创建接口需完成三步——生成订单号、扣减库存、发送通知。任何一步失败都需回滚前序操作。

def create_order(user_id, items):
    order_id = None
    try:
        # 步骤1：生成订单号（纯内存操作，无副作用）
        order_id = generate_order_id()
        
        # 步骤2：扣减库存（涉及DB事务）
        with db.transaction():
            inventory_service.decrease(items)
            
        # 步骤3：发送通知（异步，可能失败）
        notify_service.send_async(order_id, user_id)
        
    except InventoryShortageError as e:
        # 库存不足：需告知用户，且不生成订单号
        logger.warning("Inventory shortage for order %s: %s", order_id, e)
        raise HTTPException(status_code=400, detail="库存不足")
        
    except Exception as e:
        # 其他错误：记录日志，但订单号已生成，需清理
        logger.error("Unexpected error creating order %s: %s", order_id, e, exc_info=True)
        if order_id:
            cleanup_order(order_id)  # 清理已生成的订单号
        raise HTTPException(status_code=500, detail="系统繁忙")
        
    else:
        # 仅当try块无异常才执行：表示所有步骤成功！
        logger.info("Order %s created successfully", order_id)
        return {"order_id": order_id}
        
    finally:
        # 无论成功失败都执行：释放资源、关闭连接
        # 这里放db.close()或清理临时文件
        pass

关键点解析：

• else 块是 成功路径的专属通道 。它确保只有当所有可能抛异常的操作（包括 notify_service.send_async ）都成功时才进入，避免了在 try 块末尾加 return 导致 except 无法捕获后续异常的陷阱。
• finally 块是 资源守门员 。即使 except 中 raise 新异常， finally 仍会执行。这里放 db.close() 比在每个 except 里重复写更安全。
• except 按** specificity from specific to general**排序：先捕获 InventoryShortageError （业务自定义异常），再捕获 Exception 兜底，避免宽泛异常提前截获具体错误。

3.2 自定义异常：为什么 class PaymentFailedError(Exception) 比 raise Exception("payment failed") 强十倍

内置异常如 ValueError 语义太泛。当你的支付网关返回 {"code": "PAYMENT_DECLINED", "msg": "余额不足"} ，如果只抛 ValueError("payment failed") ，上层代码无法区分这是用户余额问题（应引导充值），还是签名错误（需联系运维）。自定义异常是 错误语义的载体 ：

class PaymentError(Exception):
    """所有支付相关异常的基类"""
    def __init__(self, code: str, message: str, details: dict = None):
        super().__init__(message)
        self.code = code
        self.details = details or {}

class InsufficientBalanceError(PaymentError):
    """余额不足"""
    def __init__(self, balance: float, required: float):
        msg = f"余额不足：当前{balance}元，需{required}元"
        super().__init__("INSUFFICIENT_BALANCE", msg, {"balance": balance, "required": required})

# 使用时
try:
    payment_service.charge(user_id, amount)
except InsufficientBalanceError as e:
    # 精准识别，返回特定提示
    return {"code": "INSUFFICIENT_BALANCE", "tip": "请先充值"}
except PaymentError as e:
    # 兜底处理其他支付错误
    logger.error("Payment failed: %s", e)
    return {"code": e.code, "tip": "支付失败，请重试"}

优势：

• 类型安全 ： isinstance(e, InsufficientBalanceError) 比 "INSUFFICIENT_BALANCE" in str(e) 可靠百倍；
• 结构化数据 ： e.details 可直接序列化为API响应，无需字符串解析；
• 监控友好 ：Prometheus指标可按 e.__class__.__name__ 维度统计错误率。

实操心得：自定义异常类名必须以 Error 结尾（PEP 8规范），且继承自 Exception 而非 BaseException 。避免在 __init__ 里做耗时操作（如DB查询），异常实例化应轻量。

3.3 上下文管理器与 with 语句：让资源清理不再靠“自觉”

手动写 try/finally 关文件、关数据库连接太易错。 with 语句通过 __enter__ / __exit__ 协议，把资源生命周期交给Python解释器管理。但很多人只用它开文件，却不知它能构建强大的错误隔离层：

from contextlib import contextmanager
import time

@contextmanager
def timeout(seconds: int):
    """超时上下文管理器，替代signal.alarm（不兼容Windows）"""
    start = time.time()
    try:
        yield
    except Exception as e:
        if time.time() - start > seconds:
            raise TimeoutError(f"Operation timed out after {seconds}s") from e
        raise
    if time.time() - start > seconds:
        raise TimeoutError(f"Operation timed out after {seconds}s")

# 使用
try:
    with timeout(5):
        result = slow_external_api_call()  # 若超时，抛TimeoutError
except TimeoutError as e:
    logger.warning("API call timeout: %s", e)
    # 启动降级方案：返回缓存数据
    result = get_cached_data()

更高级用法：结合 contextlib.suppress 静默特定异常（比 try: ... except SomeError: pass 更简洁）：

from contextlib import suppress

# 安全删除文件，不存在也不报错
with suppress(FileNotFoundError):
    os.remove("/tmp/temp_file.lock")

# 安全关闭socket，已关闭也不报错
with suppress(OSError):
    sock.close()

注意： suppress 只适用于 完全预期且无需处理的异常 。若需记录日志或触发告警，仍须用 try/except 。

4. 实操过程与核心环节实现：从本地脚本到云原生服务的全链路防御

4.1 脚本级防御：如何让爬虫不因一个404页面全军覆没

写爬虫时， requests.get(url) 遇到404会抛 requests.exceptions.HTTPError ，但新手常忽略 ConnectionError （DNS失败）、 Timeout （网络卡顿）、 TooManyRedirects （重定向环）。一个健壮的爬取函数应这样写：

import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

def robust_get(url: str, timeout: int = 10) -> requests.Response:
    # 配置重试策略：对5xx和网络错误重试3次，间隔1/2/4秒
    session = requests.Session()
    retry_strategy = Retry(
        total=3,
        status_forcelist=[429, 500, 502, 503, 504],  # 429是限流
        backoff_factor=1,  # 指数退避：1, 2, 4秒
        allowed_methods=["HEAD", "GET", "OPTIONS"]  # 只对安全方法重试
    )
    adapter = HTTPAdapter(max_retries=retry_strategy)
    session.mount("http://", adapter)
    session.mount("https://", adapter)
    
    try:
        response = session.get(url, timeout=timeout)
        response.raise_for_status()  # 对4xx/5xx抛HTTPError
        return response
    except requests.exceptions.Timeout:
        logger.error("Request timeout for %s", url)
        raise  # 让调用方决定是否重试
    except requests.exceptions.ConnectionError:
        logger.error("Connection failed for %s", url)
        raise
    except requests.exceptions.HTTPError as e:
        if e.response.status_code == 404:
            logger.info("Page not found: %s", url)  # 404是正常业务态
            return e.response  # 返回空响应，由业务逻辑处理
        else:
            logger.error("HTTP error %s for %s", e.response.status_code, url)
            raise
    except Exception as e:
        logger.critical("Unexpected error fetching %s: %s", url, e, exc_info=True)
        raise

# 调用示例
for url in urls:
    try:
        resp = robust_get(url)
        if resp.status_code == 200:
            parse_content(resp.text)
    except Exception as e:
        # 单个URL失败不影响整体流程
        continue

关键设计：

• 重试策略分离 ：网络层重试（ Retry ）解决瞬时故障；业务层重试（ tenacity 装饰器）解决幂等性问题；
• 404特殊处理 ：不视为错误，而是业务常态，返回响应供上层判断；
• 异常透传 ： raise 原异常，保留完整traceback，避免 raise Exception(str(e)) 丢失堆栈。

4.2 Web服务级防御：FastAPI中的异常处理金字塔

FastAPI将异常处理分为三层，每层解决不同问题：

层级	位置	处理目标	示例
路由层	@app.get() 内	业务逻辑错误，需返回HTTP状态码	raise HTTPException(status_code=404, detail="User not found")
异常处理器层	@app.exception_handler()	全局异常映射，统一响应格式	将 ValueError 转为 400 ， DatabaseError 转为 500
中间件层	@app.middleware("http")	请求生命周期错误，如JWT解析失败、请求体过大	捕获 RequestValidationError ，记录恶意请求

完整实现：

from fastapi import FastAPI, Request, HTTPException, status
from fastapi.responses import JSONResponse
from pydantic import ValidationError
import logging

app = FastAPI()

# 1. 自定义异常处理器：将所有Exception转为JSON响应
@app.exception_handler(Exception)
async def global_exception_handler(request: Request, exc: Exception):
    # 记录完整错误
    logger.error(
        "Global exception at %s %s: %s",
        request.method, request.url.path, exc,
        exc_info=True,
        extra={"request_id": request.state.request_id}  # 假设已注入request_id
    )
    return JSONResponse(
        status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
        content={"code": "INTERNAL_ERROR", "message": "系统繁忙，请稍后再试"}
    )

# 2. 专门处理Pydantic验证错误（422）
@app.exception_handler(ValidationError)
async def validation_exception_handler(request: Request, exc: ValidationError):
    logger.warning("Validation error at %s: %s", request.url.path, exc)
    return JSONResponse(
        status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
        content={"code": "VALIDATION_ERROR", "errors": exc.errors()}
    )

# 3. 中间件：捕获请求解析阶段错误（如JSON格式错误）
@app.middleware("http")
async def catch_request_errors(request: Request, call_next):
    try:
        return await call_next(request)
    except json.JSONDecodeError as e:
        logger.warning("Invalid JSON in request %s: %s", request.url.path, e)
        return JSONResponse(
            status_code=status.HTTP_400_BAD_REQUEST,
            content={"code": "INVALID_JSON", "message": "JSON格式错误"}
        )
    except Exception as e:
        # 兜底，防止中间件自身崩溃
        logger.critical("Middleware error: %s", e, exc_info=True)
        return JSONResponse(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            content={"code": "MIDDLEWARE_ERROR", "message": "系统内部错误"}
        )

# 4. 路由中使用：精准抛出业务异常
@app.get("/users/{user_id}")
async def get_user(user_id: int):
    if user_id <= 0:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="user_id must be positive integer"
        )
    user = db.get_user(user_id)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"User {user_id} not found"
        )
    return user

实操心得：异常处理器的注册顺序很重要！FastAPI按注册顺序匹配，所以把 ValidationError 处理器放在 Exception 处理器之前，否则所有验证错误都会被 Exception 捕获。

4.3 分布式任务级防御：Celery中如何让一个失败的任务不拖垮整个队列

Celery任务若未处理异常，会不断重试直至 max_retries ，而默认 max_retries=3 ，若每次重试都失败，任务会进入 failed 状态并停止。但更危险的是 任务函数内未捕获的异常导致worker进程崩溃 。正确姿势：

from celery import Celery
import logging

app = Celery('tasks')

@app.task(bind=True, autoretry_for=(ConnectionError, TimeoutError), retry_kwargs={'max_retries': 3, 'countdown': 60})
def send_notification(self, user_id: int, message: str):
    """
    bind=True: 让self成为task实例，可访问重试次数等
    autoretry_for: 自动重试指定异常类型
    """
    try:
        # 业务逻辑
        email_service.send(user_id, message)
    except EmailRateLimitError as e:
        # 速率限制：立即失败，不重试（避免雪崩）
        logger.warning("Email rate limit for user %s: %s", user_id, e)
        raise self.retry(exc=e, countdown=300)  # 5分钟后重试
    except Exception as e:
        # 其他异常：记录日志，让Celery按配置重试
        logger.error("Failed to send notification to %s: %s", user_id, e, exc_info=True)
        raise

# 全局任务失败钩子：任务最终失败时触发
@app.task_failure.connect
def task_failure_handler(sender=None, exception=None, traceback=None, **kwargs):
    logger.critical(
        "Task %s failed permanently: %s\n%s",
        sender.name, exception, traceback
    )
    # 发送告警
    alert_service.send("Celery Task Failed", str(exception))

关键配置：

• autoretry_for 指定 可重试异常 ，避免对 ValueError 等逻辑错误重试；
• retry_kwargs 中 countdown 用 指数退避 （ countdown=60 * (2 ** self.request.retries) ）；
• task_failure.connect 钩子捕获 永久失败 ，用于告警和人工介入。

5. 常见问题与排查技巧实录：那些年我们踩过的异常处理深坑

5.1 “异常消失了”： except 块里又抛异常，traceback却不见了

现象：代码里写了 except ValueError: log_error(); raise ，但日志里只看到新异常，看不到原始 ValueError 的traceback。

原因： raise 不带参数时，会重新抛出 当前异常上下文 ，但若 log_error() 里又抛了异常，就会覆盖原始异常。正确写法是 raise 带 from ：

try:
    risky_operation()
except ValueError as e:
    logger.error("Value error occurred: %s", e)
    # 错误：直接raise会丢失原始traceback
    # raise
    
    # 正确：用raise ... from ... 保留因果链
    raise RuntimeError("Failed to process input") from e

效果：日志中会显示：

RuntimeError: Failed to process input
...
The above exception was the direct cause of the following exception:
...
ValueError: invalid literal for int() with base 10: 'abc'

排查技巧：用 sys.exc_info() 获取当前异常三元组（type, value, traceback）， traceback.print_exception(*sys.exc_info()) 可手动打印完整堆栈。

5.2 “日志爆炸”：循环导入导致 ImportError 引发无限递归

现象：服务启动时CPU飙升100%，日志刷屏 ImportError: cannot import name 'xxx' ，且错误位置在 __init__.py 。

根因：模块A导入模块B，模块B又导入模块A，形成循环。当A的 __init__.py 执行到一半，B尝试导入A的某个函数时，A尚未加载完成，抛 ImportError 。而若A的 __init__.py 里有 try/except ImportError 且 except 块又尝试导入B，则陷入死循环。

解决方案：

• 重构依赖 ：将共享代码抽到独立模块C，A和B都导入C；
• 延迟导入 ：在函数内部导入（ def func(): import requests ），而非模块顶层；
• 使用 importlib.import_module() 动态导入 ，配合 try/except 。

5.3 “告警失灵”：为什么Prometheus监控不到你的异常？

现象：代码里 logger.error("DB error", exc_info=True) ，但Prometheus的 python_exceptions_total 指标没变化。

原因： python_exceptions_total 是 prometheus_client 库的默认指标，它只统计 被 prometheus_client.Counter 显式计数的异常 ，不会自动捕获所有 logger.error 。你需要主动埋点：

from prometheus_client import Counter

EXCEPTION_COUNTER = Counter(
    'myapp_exceptions_total',
    'Total number of exceptions',
    ['type', 'handler']  # 按异常类型和处理位置标签
)

@app.exception_handler(DatabaseError)
async def db_exception_handler(request: Request, exc: DatabaseError):
    EXCEPTION_COUNTER.labels(type='DatabaseError', handler='api').inc()
    logger.error("Database error: %s", exc)
    return JSONResponse(...)

速查表：常见异常排查命令

问题现象	快速定位命令	说明
程序崩溃但无日志	python -u script.py 2>&1 | grep -i "exception|error"	-u 禁用输出缓冲，确保实时看到错误
traceback太长难定位	python -c "import traceback; traceback.print_tb(traceback.extract_stack()[-1])"	打印最后一帧，快速定位问题行
想知道哪个包抛了异常	pip show requests + 查看其 __init__.py 中的 raise 语句	第三方库异常源码就在那
生产环境无法debug	import pdb; pdb.set_trace() 替换为 breakpoint() （Python 3.7+）	breakpoint() 会读取 PYTHONBREAKPOINT 环境变量，可设为 0 禁用

5.4 “测试失效”：单元测试里 assertRaises 为何不生效？

现象：写 self.assertRaises(ValueError, int, "abc") ，但测试通过，而实际运行时 int("abc") 确实抛 ValueError 。

原因： assertRaises 要求 被测函数必须在测试方法内直接调用 。若你写成：

def test_bad_input(self):
    result = int("abc")  # 错误：这里就抛异常了，assertRaises没机会捕获
    self.assertRaises(ValueError, result)  # 永远不会执行到这里

正确写法：

def test_bad_input(self):
    # 方式1：传入函数和参数
    self.assertRaises(ValueError, int, "abc")
    
    # 方式2：用with语句（推荐，可检查异常属性）
    with self.assertRaises(ValueError) as cm:
        int("abc")
    self.assertIn("invalid literal", str(cm.exception))

实操心得：用 pytest 时， with pytest.raises(ValueError): 更简洁；对异步函数，用 pytest-asyncio 的 await pytest.raises(...) 。

6. 工具链与进阶实践：让异常处理从手工劳动变成自动化工程

6.1 静态分析：用 pylint 和 mypy 在编码阶段拦截异常隐患

pylint 能检测未处理的异常（ broad-except 警告）和空 except 块：

pip install pylint
pylint --enable=broad-except,empty-except mymodule.py

mypy （Python类型检查器）可发现潜在的 None 解引用错误，这类错误常在运行时才抛 AttributeError ：

def get_user_name(user_id: int) -> str:
    user = db.find_user(user_id)  # 类型声明：Optional[User]
    return user.name  # mypy报错：可能为None！

# 修复后
def get_user_name(user_id: int) -> str:
    user = db.find_user(user_id)
    if user is None:
        raise UserNotFoundError(f"User {user_id} not found")
    return user.name

6.2 动态追踪：用 sys.settrace 实现异常调用链可视化

想搞清一个异常从哪来、经过哪些函数？ sys.settrace 可埋点：

import sys
import traceback

def trace_calls(frame, event, arg):
    if event == 'exception':
        exc_type, exc_value, exc_traceback = arg
        print(f"Exception {exc_type.__name__}: {exc_value}")
        # 打印调用栈前3层
        for line in traceback.format_tb(exc_traceback)[-3:]:
            print(line.rstrip())
    return trace_calls

# 启用追踪
sys.settrace(trace_calls)
# 运行你的代码...
sys.settrace(None)  # 关闭

6.3 监控告警：ELK+Prometheus异常模式识别

在ELK中，用Logstash过滤出 ERROR 日志，提取 exception_type 字段：

filter {
  if [message] =~ /Exception|Error/ {
    grok {
      match => { "message" => "%{TIMESTAMP_ISO8601:timestamp} %{LOGLEVEL:level}.*?%{JAVACLASS:exception_type}:" }
    }
  }
}

在Kibana中创建可视化：按 exception_type 聚合，设置告警——当 DatabaseTimeoutError 5分钟内超过10次，触发PagerDuty。

最后分享一个小技巧：在所有 except 块第一行加 logger.debug("Entering except block for %s", type(e).__name__) 。这看似多余，但在复杂嵌套异常中，它能帮你确认代码是否真的走到了你认为的 except 分支——我靠这行日志定位过三次“明明写了except却没生效”的诡异问题。