本文介绍如何在已使用 QueueHandler + RotatingFileHandler 实现异步轮转日志的基础上，无缝集成 SMTPHandler，使 logger.warning()、logger.error() 或 logger.critical() 触发时自动发送邮件告警。
本文介绍如何在已使用 queuehandler + rotatingfilehandler 实现异步轮转日志的基础上，无缝集成 smtphandler，使 `logger.warning()`、`logger.error()` 或 `logger.critical()` 触发时自动发送邮件告警。
在生产环境中，仅将日志写入文件往往不足以保障快速响应——尤其当关键异常发生时，开发者需要即时感知。Python 标准库的 logging 模块支持多处理器（handler）并行工作，而 QueueListener 正是协调异步日志分发的理想中枢。本文将基于您已有的轮转日志架构，扩展一个高可靠、低侵入的邮件告警能力。
Secure Python Code Sandbox
下载
安全 Python 代码沙箱：在隔离的沙箱中执行任意 Python 代码，预装 requests、NumPy、pandas 库。返回 stdout、stderr。
✅ 核心实现要点
-
复用现有队列机制：无需额外线程或修改主日志流，直接将
SMTPHandler注册到QueueListener的 handler 列表中； -
精准控制触发级别：通过
mailHandler.setLevel(logging.WARNING)限定仅WARNING及以上（即WARNING/ERROR/CRITICAL）触发邮件； -
启用级别过滤开关：必须设置
respect_handler_level=True（默认为False），否则所有日志（包括INFO）都会被转发给SMTPHandler； -
避免弃用方法：
logger.warn()已被弃用，请统一使用logger.warning()（Python 3.3+ 强制要求）。
? 配置 SMTPHandler 示例（含安全建议）
import logging
from logging.handlers import SMTPHandler, RotatingFileHandler, QueueListener
from queue import Queue
def initialize(
app_name="app_name",
log_level=logging.DEBUG,
log_size=10 * 1024 * 1024,
num_files=10,
# 邮件配置（请按实际替换）
mailhost=("smtp.gmail.com", 587),
fromaddr="your_app@example.com",
toaddrs=["admin@example.com"],
subject="[ALERT] Your App Log Warning",
credentials=("your_email@gmail.com", "your_app_password_or_app_token"),
secure=() # 使用 TLS；若需 SSL，设为 None 或传入空元组（取决于 SMTP 服务）
):
logger = logging.getLogger(__name__)
logger.setLevel(log_level)
logger.propagate = False # 防止日志向上冒泡至 root logger 导致重复输出
# 1. 创建队列与 QueueHandler
log_queue = Queue()
queue_handler = logging.handlers.QueueHandler(log_queue)
logger.addHandler(queue_handler)
# 2. 创建 RotatingFileHandler（JSON 格式化）
log_path = get_log_path(app_name) # 您已实现的路径生成函数
rotating_file_handler = RotatingFileHandler(
log_path, maxBytes=log_size, backupCount=num_files
)
rotating_file_handler.setFormatter(JSONLogFormatter()) # 您自定义的 JSON 格式器
# 3. 创建 SMTPHandler（仅处理 WARNING+）
mail_handler = SMTPHandler(
mailhost=mailhost,
fromaddr=fromaddr,
toaddrs=toaddrs,
subject=subject,
credentials=credentials,
secure=secure
)
mail_handler.setLevel(logging.WARNING) # 关键：设定触发级别
mail_handler.setFormatter(logging.Formatter(
'%(asctime)s - %(name)s - %(levelname)s - %(message)s'
))
# 4. 启动 QueueListener，同时监听多个 handler，并启用级别过滤
queue_listener = QueueListener(
log_queue,
rotating_file_handler,
mail_handler,
respect_handler_level=True # ⚠️ 必须设为 True！
)
return logger, queue_listener⚠️ 注意事项与最佳实践
-
凭证安全：切勿硬编码邮箱密码。推荐使用环境变量（如
os.getenv("SMTP_PASSWORD")）或密钥管理服务； - Gmail 用户注意：需开启「App Password」（两步验证后生成），而非账户密码；或使用 OAuth2（更复杂但更安全）；
-
网络可靠性：
SMTPHandler默认同步发送邮件，若 SMTP 服务不可达，可能阻塞QueueListener线程。如需更高可用性，可封装为异步任务（如结合threading.Thread或concurrent.futures），但会增加复杂度； - 邮件频控：生产环境建议添加限流逻辑（例如每分钟最多发 5 封），避免因高频错误导致邮箱被封或告警疲劳；
-
测试验证：务必在
_test()中显式调用logger.warning()和logger.error()进行端到端测试，确认邮件真实到达。
✅ 最终验证示例
def _test():
logger, listener = initialize(
toaddrs=["your_real_email@example.com"],
credentials=("your_email@gmail.com", "your_app_token")
)
listener.start()
try:
logger.info("This goes only to file (no email).")
logger.warning("This triggers an email alert.")
logger.error("Critical issue detected — email sent immediately.")
finally:
listener.stop()
listener.join() # 确保队列中剩余日志被处理完毕通过以上配置，您在不干扰原有高性能日志写入的前提下，获得了及时、可控、可审计的异常告警能力——真正实现“日志归档”与“事件响应”的双轨协同。
