You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

Python避免日志泛滥方案:忽略重复错误及logging模块内置功能咨询

避免日志泛滥:重复错误的限流方案

Python logging模块的内置方案?

Python标准库的logging模块本身没有原生支持重复日志的时间限流功能,但可以通过扩展它的Filter或Handler组件来实现,这是更贴合logging模块设计思路的方式。

自定义实现方案

方案1:基于缓存的简单限流(适合小型脚本)

在日志函数中维护一个缓存字典,记录每种错误的最后日志时间,每次日志前检查时间间隔是否满足要求:

import time
import logging

# 初始化logging基础配置
logging.basicConfig(level=logging.ERROR)

# 缓存:键为错误标识(类型+消息),值为最后日志时间戳
_last_log_times = {}
# 限流间隔:60秒(1分钟)
LOG_INTERVAL = 60

def log_error(error):
    # 生成唯一错误标识:结合异常类型名和错误消息
    error_key = (type(error).__name__, str(error))
    current_time = time.time()
    
    # 检查是否满足日志条件
    if error_key not in _last_log_times or (current_time - _last_log_times[error_key]) >= LOG_INTERVAL:
        logging.error(str(error), exc_info=True)
        _last_log_times[error_key] = current_time

def do_something_that_fails():
    return 1/0

while 1:
    try:
        do_something_that_fails()
    except Exception as error:
        log_error(error)
    time.sleep(5)

方案2:自定义logging Filter(贴合logging生态)

如果项目已在使用logging的多Handler/Logger结构,推荐用自定义Filter实现限流,所有挂载该Filter的Logger都会自动生效:

import time
import logging

class RateLimitFilter(logging.Filter):
    def __init__(self, interval=60):
        super().__init__()
        self.interval = interval
        self._last_log_times = {}

    def filter(self, record):
        # 基于日志的异常信息或消息生成唯一标识
        if record.exc_info:
            exc_type, exc_obj, _ = record.exc_info
            error_key = (exc_type.__name__, str(exc_obj))
        else:
            error_key = record.message
        
        current_time = time.time()
        if error_key not in self._last_log_times or (current_time - self._last_log_times[error_key]) >= self.interval:
            self._last_log_times[error_key] = current_time
            return True  # 允许日志输出
        return False  # 拒绝日志输出

# 配置logging
logger = logging.getLogger(__name__)
logger.setLevel(logging.ERROR)
handler = logging.StreamHandler()
handler.addFilter(RateLimitFilter(interval=60))
logger.addHandler(handler)

def log_error(error):
    logger.error(str(error), exc_info=True)

def do_something_that_fails():
    return 1/0

while 1:
    try:
        do_something_that_fails()
    except Exception as error:
        log_error(error)
    time.sleep(5)

比较异常的恰当方法

直接用==比较两个Exception实例通常不可靠——即使是同一错误类型、相同消息的异常,也是不同的实例。正确的比较方式:

  • 比较异常类型:type(error1) == type(error2)
  • 比较异常核心信息:比如str(error1) == str(error2),或直接比较error1.args == error2.args(args是异常构造时传入的参数,更底层准确)
  • 自定义异常可额外比较其专属属性字段

上面的代码中,我们用(type(error).__name__, str(error))作为错误标识,兼顾类型和消息,足以区分大多数重复错误场景。

内容的提问来源于stack exchange,提问作者Mr. B.

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.08.09 11:50:50