from tenacity import retry, stop_after_attempt, wait_fixed, before_log

# 基础重试结构
@retry(
    stop=stop_after_attempt(3),      # 最多重试3次
    wait=wait_fixed(2),               # 每次间隔2秒
    before=before_log(logger, 'WARNING')  # 重试前记录日志
)
def api_call():
    # 可能失败的逻辑

通过Tenacity的回调钩子，在重试发生时自动记录日志，无需修改原函数逻辑。

import logging
from tenacity import retry, stop_after_attempt, wait_fixed, RetryCallState

# 配置日志记录器
logger = logging.getLogger(__name__)

def custom_before_log(retry_state: RetryCallState):
    """自定义重试前日志钩子"""
    if retry_state.attempt_number > 1:
        logger.warning(
            f"函数 {retry_state.fn.__name__} 第{retry_state.attempt_number}次重试"
        )

@retry(
    stop=stop_after_attempt(3),
    wait=wait_fixed(1),
    before=custom_before_log  # 挂载钩子函数
)
def login(user, password):
    # 模拟登录操作
    if random.random() > 0.4:
        raise ConnectionError("认证服务不可用")
    return "登录成功"

钩子函数核心参数解析

参数	说明	使用场景
attempt_number	当前重试次数	显示重试进度
fn.__name__	函数名称	定位问题函数
outcome	执行结果对象	获取异常详情

优点：

非侵入式：无需修改原函数代码
配置简单：只需添加before参数
低耦合：重试逻辑与业务逻辑分离

通过自定义装饰器封装重试逻辑，可记录更详细的上下文信息（如函数参数）。

import inspect
from functools import wraps
from tenacity import retry, stop_after_attempt, wait_fixed

def retry_with_logging(stop_max=3, wait_seconds=2):
    """带日志记录的自定义重试装饰器"""
    def decorator(func):
        @wraps(func)
        @retry(
            stop=stop_after_attempt(stop_max),
            wait=wait_fixed(wait_seconds)
        )
        def wrapper(*args, **kwargs):
            try:
                return func(*args, **kwargs)
            except Exception as e:
                # 动态获取函数参数
                sig = inspect.signature(func)
                bound_args = sig.bind(*args, **kwargs)
                args_str = ", ".join(
                    f"{k}={v!r}" for k,v in bound_args.arguments.items()
                )
                
                # 记录详细日志
                logger.warning(
                    f"函数 {func.__name__}({args_str}) "
                    f"第{wrapper.retry_state.attempt_number}次失败: {e}"
                )
                raise e
        return wrapper
    return decorator

# 使用示例
@retry_with_logging(stop_max=4, wait_seconds=3)
def process_data(data_id, priority='high'):
    # 数据处理逻辑
    if random.random() > 0.3:
        raise ResourceWarning("资源暂时不可用")

关键技术解析

1.动态参数捕获：

inspect.signature(func)  # 获取函数签名
sig.bind(*args, **kwargs)  # 绑定实际参数

2.重试状态追踪：

wrapper.retry_state.attempt_number  # 当前重试次数

3.异常上下文记录：

将异常对象e与参数值一起记录，便于复现问题

优点：

参数可视化：记录调用时的具体参数值
高度定制化：可扩展结果检查、异常过滤等逻辑
复用性强：一次封装，多处使用

特性钩子函数法装饰器封装法

实现复杂度 ⭐（简单） ⭐⭐⭐（中等）

日志详细度 ⭐⭐ ⭐⭐⭐⭐

侵入性无需添加装饰器

参数记录不支持完整记录

适用场景快速集成关键业务逻辑

选型建议：

选择钩子函数法当：只需基础重试次数记录、希望零侵入现有代码、快速原型开发

选择装饰器封装法当：需要排查参数相关错误、处理核心业务逻辑、需要复用重试配置

特性	钩子函数法	装饰器封装法
实现复杂度	⭐（简单）	⭐⭐⭐（中等）
日志详细度	⭐⭐	⭐⭐⭐⭐
侵入性	无	需添加装饰器
参数记录	不支持	完整记录
适用场景	快速集成	关键业务逻辑

避免高频重试导致服务雪崩：

from tenacity import wait_exponential

@retry(wait=wait_exponential(multiplier=1, max=60))
def api_call():
    # 等待时间：1s → 2s → 4s → ... → 60s

只重试特定异常类型：

from tenacity import retry_if_exception_type

@retry(retry=retry_if_exception_type((TimeoutError, ConnectionError)))

@retry(
    stop=(stop_after_attempt(5) | stop_after_delay(30)),  # 5次或30秒后停止
    wait=wait_random(min=1, max=10),                     # 随机等待1-10秒
    after=release_resource  # 重试结束后释放资源
)

1.避免无限重试

始终设置stop条件（次数或时间上限），防止死循环

2.区分可重试错误

仅重试临时性错误（如网络超时），跳过业务逻辑错误（如密码错误）

3.关键操作添加回调

在after回调中关闭连接、释放锁等资源

def cleanup(retry_state):
    if retry_state.outcome.failed:
        close_db_connection()

4.生产环境监控

结合Sentry等工具对重试事件报警，配置示例：

from sentry_sdk import capture_message

def alert_on_retry(retry_state):
    if retry_state.attempt_number > 3:
        capture_message(f"高频重试: {retry_state.fn.__name__}")

场景配置方案成功率提升

API调用指数退避+3次重试 78% → 97%

数据库操作固定间隔+异常过滤 82% → 99.5%

文件上传随机等待+参数日志 65% → 93%

实测案例：某支付系统接入重试机制后，在AWS区域性网络故障期间，支付失败率从18%降至0.7%。

场景	配置方案	成功率提升
API调用	指数退避+3次重试	78% → 97%
数据库操作	固定间隔+异常过滤	82% → 99.5%
文件上传	随机等待+参数日志	65% → 93%

1.识别可重试操作：数据库/API/文件等可能临时失败的操作

2.选择记录方案：

快速集成 → 钩子函数法
深度追踪 → 装饰器封装法

3.配置策略：

# 最佳实践模板
@retry(
    stop=stop_after_attempt(4),
    wait=wait_exponential(max=30),
    before=log_attempt_number,
    retry=retry_if_exception_type(TransientError)
)

重试机制不是万能药，但合理使用能显著提升系统韧性。当你的代码再次面对网络波动或服务抖动时，它将不再脆弱崩溃，而是优雅地记录问题、智能重试，最终完成使命——这正是专业级应用的标志。

到此这篇关于深入解析Python高效记录重试日志的两种方法的文章就介绍到这了,更多相关Python日志重试内容请搜索风君子博客以前的文章或继续浏览下面的相关文章希望大家以后多多支持风君子博客！

您可能感兴趣的文章:

python使用logging模块记录日志的操作
Python 日志记录模块的综合指南
Python中更优雅的日志记录方案详解
Python 程序员必须掌握的日志记录
Python retrying 重试机制详解
Python retrying 重试机制的使用方法

华夏福临门盛世加强版怎么样？产品责任是重中之重

国寿福庆典版值得买吗？附投保建议

理财中低风险是几级？中低风险理财会亏吗

瑞泰泰安心防癌险优缺点是什么？这份分析不容错过

2020最新自助免费申请Office365教育版，免费5TOneDrive云盘详细图文教程

MacBook怎么绕过BootCamp安装Win10双系统？

国内知名的前端博客

域名为什么会被墙？如何检测域名是否被墙？域名被墙如何处理？

最新的windows xp sp3产品密匙 xp sp3序列号 xp sp3激活码

深入解析Python高效记录重试日志​​的两种方法

目录

避免高频重试导致服务雪崩： from tenacity import wait_exponential @retry(wait=wait_exponential(multiplier=1, max=60)) def api_call(): # 等待时间：1s → 2s → 4s → ... → 60s

只重试特定异常类型： from tenacity import retry_if_exception_type @retry(retry=retry_if_exception_type((TimeoutError, ConnectionError)))

@retry( stop=(stop_after_attempt(5) | stop_after_delay(30)), # 5次或30秒后停止 wait=wait_random(min=1, max=10), # 随机等待1-10秒 after=release_resource # 重试结束后释放资源 )

More posts

站内搜索

标签云

热门文章

友情链接

深入解析Python高效记录重试日志的两种方法

避免高频重试导致服务雪崩：

from tenacity import wait_exponential @retry(wait=wait_exponential(multiplier=1, max=60)) def api_call(): # 等待时间：1s → 2s → 4s → ... → 60s

只重试特定异常类型：

from tenacity import retry_if_exception_type @retry(retry=retry_if_exception_type((TimeoutError, ConnectionError)))