Python 如何捕获日志中的异常并发送通知

舞姬之光

发布时间：2025-09-21 08:24:01

825人浏览过

来源于php中文网

原创

通过自定义logging.Handler实现异常实时通知，结合限流、异步发送与上下文丰富等策略，可高效捕获并推送Python应用中的错误信息至Slack、钉钉等平台，提升生产环境问题响应速度。

python 如何捕获日志中的异常并发送通知

在Python项目中，捕获日志中的异常并及时发送通知，本质上就是利用Python强大的

logging

模块，结合自定义的日志处理器（Handler），在特定的日志级别（通常是

ERROR

或

CRITICAL

）触发时，将异常信息通过各种通道（如邮件、即时通讯工具API、Webhook等）推送出去。这能帮助我们第一时间发现并响应生产环境中的问题，而不是被动地等待用户反馈或定时查阅日志文件。

解决方案

要实现这一点，我们需要几个核心步骤。首先，配置好

logging

模块，确保它能正确捕获到异常信息。这通常意味着在

logging.error()

或

logging.exception()

调用时，传递

exc_info=True

参数，或者直接使用

logging.exception()

，它会自动包含当前异常的详细信息。

接下来，关键在于创建一个或多个自定义的日志处理器。Python标准库自带了

logging.handlers.SMTPHandler

用于发送邮件，但对于更现代的通知方式（如Slack、钉钉、企业微信、PagerDuty等），我们通常需要编写一个继承自

logging.Handler

的类。

基础日志配置与异常捕获示例：

立即学习“Python免费学习笔记（深入）”；

import logging
import traceback
import requests # 假设用于发送HTTP请求到通知服务
import json
import os # 用于获取环境变量，避免硬编码敏感信息

# 配置基础日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)

# 定义一个简单的自定义通知Handler
class CustomNotificationHandler(logging.Handler):
    def __init__(self, webhook_url, level=logging.ERROR):
        super().__init__(level)
        self.webhook_url = webhook_url
        self.session = requests.Session() # 使用session保持连接，提高效率

    def emit(self, record):
        # 过滤掉非异常的日志，或者根据需求只处理特定级别的日志
        if not record.exc_info and record.levelno < logging.ERROR:
            return

        # 格式化日志信息，包括异常堆栈
        message = self.format(record)

        # 准备发送到通知服务的payload
        # 这里以一个简单的Webhook为例，实际可能需要根据服务调整
        payload = {
            "text": f"? **生产环境异常告警** ?\n\n**应用:** MyAwesomeApp\n**级别:** {record.levelname}\n**消息:** {record.message}\n**时间:** {self.formatTime(record, '%Y-%m-%d %H:%M:%S')}\n\n**详细信息:**\n```\n{message}\n```"
        }

        try:
            # 异步发送通知是更好的实践，这里为简化直接发送
            # 生产环境建议使用线程池、Celery等异步任务队列
            response = self.session.post(self.webhook_url, json=payload, timeout=5)
            response.raise_for_status() # 检查HTTP请求是否成功
        except requests.exceptions.RequestException as e:
            # 如果通知发送失败，我们应该记录下来，但不能再次触发通知循环
            print(f"Failed to send notification: {e}")
            # 这里可以考虑将失败的通知信息记录到另一个更稳定的地方，如数据库或专门的告警系统
        except Exception as e:
            print(f"Unexpected error in CustomNotificationHandler: {e}")

# 获取Webhook URL，通常从环境变量或配置文件读取
# SLACK_WEBHOOK_URL = os.getenv("SLACK_WEBHOOK_URL") 
# 示例用一个虚拟的URL
SLACK_WEBHOOK_URL = "https://hooks.slack.com/services/T00000000/B00000000/XXXXXXXXXXXXXXXXXXXXXXXX" 

if SLACK_WEBHOOK_URL:
    notification_handler = CustomNotificationHandler(SLACK_WEBHOOK_URL, level=logging.ERROR)
    # 给通知处理器设置一个更简洁的Formatter，因为我们已经在payload中格式化了大部分信息
    notification_handler.setFormatter(logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s', datefmt='%Y-%m-%d %H:%M:%S'))

    # 将自定义的Handler添加到root logger
    logging.getLogger().addHandler(notification_handler)
else:
    logging.warning("SLACK_WEBHOOK_URL not set, notification handler will not be active.")

# 模拟一个会抛出异常的函数
def problematic_function(a, b):
    return a / b

# 捕获并记录异常
try:
    result = problematic_function(10, 0)
    print(result)
except ZeroDivisionError as e:
    logging.error("An error occurred during division.", exc_info=True) # exc_info=True是关键
    # 或者直接使用 logging.exception()
    # logging.exception("Another error occurred during division.")

print("\n--- 模拟其他日志 ---")
logging.info("This is an informational message.")
logging.warning("This is a warning, but won't trigger notification by default.")

这段代码展示了一个基本的框架。

CustomNotificationHandler

是核心，它重写了

emit

方法来处理日志记录。当一个日志记录被处理时，

emit

方法会检查其级别和是否包含异常信息，然后构建一个适合目标通知服务的payload并发送。

为什么我们需要对Python日志中的异常进行实时通知？

对我来说，实时通知的价值体现在“主动”二字。传统的做法是部署服务，然后定期或在用户抱怨时去查看日志文件。这种方式效率低下且反应滞后。想象一下，如果一个核心功能在凌晨两点因为某个边缘条件崩溃了，没有实时通知，你可能要等到第二天上班，甚至更晚，才能发现问题。这期间，用户体验受损，业务损失也在悄悄累积。

Digram

让Figma更好用的AI神器

下载

有了实时通知，一旦异常发生，相关负责人（比如我，或者我的团队成员）就能立刻收到消息。这就像给系统装了一个“健康警报器”。它不仅能帮助我们更快地定位问题、减少停机时间，还能让我们对系统的健康状况有一个更直观、更及时的感知。有时候，一些看似不严重的错误，如果频繁出现，可能预示着潜在的系统瓶颈或设计缺陷，实时通知能让我们尽早注意到这些“小信号”，避免它们演变成“大事故”。它把被动救火变成了主动预防，这在任何生产环境中都是至关重要的。

除了邮件，还有哪些高效的异常通知方式及如何实现？

邮件通知虽然经典，但在即时性、团队协作和信息聚合方面，它已经显得有些力不从心了。我个人更倾向于使用那些能直接集成到团队日常工作流中的工具。

Slack/Microsoft Teams/钉钉/企业微信： 这些是目前最流行的团队协作工具。它们都提供了Webhook接口，允许你通过发送HTTP POST请求来发布消息到指定的频道。
- 实现方式： 大同小异，都是构建一个JSON格式的payload，然后用Python的
```
requests
```
  库发送到Webhook URL。Payload的结构会因平台而异，通常包含文本内容、可选的Markdown格式、附件、按钮等。
- 优点： 消息即时，团队成员都能看到，方便讨论和协作；支持富文本和交互式消息，可以包含更多上下文信息。
PagerDuty/Opsgenie： 对于需要严格On-Call轮值和事件管理的企业，这些是专业的事件响应平台。它们提供更高级的API，可以创建事件、触发告警、管理告警升级策略等。
- 实现方式： 通常会有官方或社区提供的Python SDK，或者直接调用其REST API。集成会比简单的Webhook复杂一些，但功能也更强大，适合大型、高可用性要求的系统。
- 优点： 专业的告警管理、轮值、升级策略、电话/短信通知，确保关键告警不会被遗漏。
自定义Webhooks： 如果你的团队内部有自研的告警平台，或者需要将通知转发到其他第三方系统，Webhook是最灵活的选择。
- 实现方式： 你的
```
CustomNotificationHandler
```
  只需要知道目标Webhook的URL和预期的payload格式，然后发送HTTP请求即可。这种方式的灵活性在于你可以完全控制消息内容和接收方的处理逻辑。

以Slack为例，一个简单的

CustomNotificationHandler

在

emit

方法中可以这样构建payload：

# ... (CustomNotificationHandler的init方法) ...

    def emit(self, record):
        # ... (过滤逻辑) ...

        # 格式化堆栈信息
        exc_text = ""
        if record.exc_info:
            exc_text = "".join(traceback.format_exception(*record.exc_info))

        # 针对Slack的Payload
        slack_message_blocks = [
            {
                "type": "header",
                "text": {
                    "type": "plain_text",
                    "text": f"? 异常告警: {record.levelname} ?"
                }
            },
            {
                "type": "section",
                "fields": [
                    { "type": "mrkdwn", "text": f"*应用:* MyAwesomeApp" },
                    { "type": "mrkdwn", "text": f"*时间:* {self.formatTime(record, '%Y-%m-%d %H:%M:%S')}" },
                    { "type": "mrkdwn", "text": f"*级别:* {record.levelname}" },
                    { "type": "mrkdwn", "text": f"*消息:* {record.message}" }
                ]
            }
        ]

        if exc_text:
            slack_message_blocks.append({
                "type": "section",
                "text": {
                    "type": "mrkdwn",
                    "text": "*堆栈信息:*\n```\n" + exc_text + "\n```"
                }
            })

        payload = {
            "blocks": slack_message_blocks
        }

        try:
            response = self.session.post(self.webhook_url, json=payload, timeout=5)
            response.raise_for_status()
        except requests.exceptions.RequestException as e:
            print(f"Failed to send Slack notification: {e}")

这样，消息在Slack中会以更美观、结构化的形式展现，便于阅读和理解。

在生产环境中，处理日志异常通知有哪些最佳实践和注意事项？

在生产环境里，异常通知的实现远不止“能发出去”那么简单，还需要考虑很多细节，否则可能适得其反，比如造成“告警疲劳”。

限流与去重（Rate Limiting & Deduplication）： 这是我首先会考虑的。如果一个异常在短时间内反复出现，你肯定不希望收到成百上千条重复的通知。
- 实现： 可以使用一个简单的内存缓存（如Python的
```
functools.lru_cache
```
  或一个字典）来记录某个异常（基于错误类型、消息摘要或堆栈哈希）在最近一段时间内是否已经通知过。如果已通知，则在一定冷却时间内不再发送。更复杂的系统可能会用Redis来做分布式限流。
- 注意： 限流时间不能太长，否则可能错过问题恢复后的再次恶化。
异步发送： 通知发送通常涉及网络请求，这可能会阻塞主应用程序的执行。在生产环境中，这几乎是不可接受的。
- 实现： 将通知发送逻辑放入一个独立的线程（
```
threading
```
  模块）、进程（
```
multiprocessing
```
  模块）或更健壮的异步任务队列（如Celery配合Redis/RabbitMQ）。这样，即使通知服务暂时不可用，也不会影响主应用的性能。
丰富上下文信息： 收到告警时，如果只知道“某个地方出错了”，那调试起来会非常痛苦。
- 包含信息： 除了堆栈信息，尽可能包含请求ID、用户ID、当前环境（开发/测试/生产）、服务名称、服务器IP、请求参数等。这些信息能极大地加速问题定位。
告警分级与路由： 并非所有错误都需要立即通知到所有人。
- 分级： 将错误分为
```
WARNING
```
  、
```
ERROR
```
  、
```
CRITICAL
```
  等。
```
WARNING
```
  可能只需要记录到日志文件，
```
ERROR
```
  可能通知到开发组的日常频道，而
```
CRITICAL
```
  则需要触发On-Call机制，通知到值班人员的手机。
- 路由： 根据日志级别、错误类型、甚至错误发生的频率，将通知发送到不同的渠道或不同的团队。
错误自愈与告警抑制： 有些错误是短暂的、可恢复的（比如网络瞬断），系统可能在几次重试后就能自行恢复。
- 策略： 对于这类错误，可以考虑在多次连续失败后才触发告警，或者在系统成功自愈后发送一个“问题已解决”的通知，避免无谓的告警。
安全性： 通知内容中避免包含敏感信息，如用户密码、API密钥等。Webhook URL也应妥善保管，避免泄露。
可观测性（Observability）： 除了通知，将异常数据发送到APM（应用性能管理）工具或日志聚合平台（如ELK Stack、Grafana Loki）进行统一分析和可视化，也是非常有价值的。这能帮助你发现异常的趋势、关联性，进行更深层次的故障排除。