Tenacity —— 一个简单易用的 Python 重试库

881 words5 min read

Tenacity 是一个基于 Apache 2.0 许可证的 Python 通用重试库,旨在为函数或代码块提供灵活、可配置的重试支持。它源自 retrying 项目,在兼容性和功能性方面进行了大量改进,适用于需要增强健壮性、应对间歇性故障的各类场景。

核心价值与适用场景

Tenacity 通过装饰器或上下文管理器,为任意函数或代码块提供可定制的重试策略,有效提升代码的容错能力。常见应用包括网络通信、远程服务调用、分布式任务处理等,尤其适用于易受外部环境影响、偶发错误频发的场景。

基本用法

Tenacity 的最基本用法是通过 @retry 装饰器,将重试逻辑应用于目标函数。当函数抛出异常时,Tenacity 会自动按照预设策略进行重试,直至成功或达到终止条件。

import random
from tenacity import retry

@retry
def do_something_unreliable():
    if random.randint(0, 10) > 1:
        raise IOError("操作失败")
    return "成功"

重试策略详解

Tenacity 支持多种重试终止和等待策略,用户可根据实际需求灵活配置。

停止条件

可通过 stop 参数灵活设定重试终止规则,如最大重试次数、最长重试时长,或二者结合。

from tenacity import retry, stop_after_attempt, stop_after_delay

@retry(stop=stop_after_attempt(5))  # 最多重试5次
def func1():
    ...

@retry(stop=stop_after_delay(10))  # 最长重试10秒
def func2():
    ...

等待策略

Tenacity 提供多种等待策略,包括固定间隔、随机间隔、指数退避等。

from tenacity import wait_fixed, wait_random, wait_exponential

@retry(wait=wait_fixed(2))  # 每次重试间隔2秒
def func3():
    ...

@retry(wait=wait_random(min=1, max=3))  # 间隔1~3秒随机
def func4():
    ...

@retry(wait=wait_exponential(multiplier=1, min=4, max=10))  # 指数退避
def func5():
    ...

重试条件

可通过 retry 参数指定需重试的异常类型、返回结果或自定义判断逻辑。

from tenacity import retry_if_exception_type, retry_if_result

@retry(retry=retry_if_exception_type(IOError))
def func7():
    ...

def is_none(value):
    return value is None

@retry(retry=retry_if_result(is_none))
def func8():
    ...

进阶特性

回调与日志

Tenacity 支持在重试前后、等待前等关键节点执行自定义回调,便于记录日志或执行特定操作。

from tenacity import before_log, after_log, before_sleep_log
import logging

logger = logging.getLogger(__name__)

@retry(stop=stop_after_attempt(3), before=before_log(logger, logging.INFO))
def func8():
    ...

统计信息与自定义回调

可通过函数的 retry.statistics 属性获取重试统计信息。自定义回调函数可基于 retry_state 对象实现更复杂的控制逻辑。

异步支持

Tenacity 原生支持 asyncio、Tornado 等异步框架,适用于现代异步编程场景。

from tenacity import retry

@retry
async def my_async_function():
    ...

代码块重试与运行时配置

除了函数装饰器,Tenacity 还支持对任意代码块进行重试,便于细粒度控制。

from tenacity import Retrying, stop_after_attempt

for attempt in Retrying(stop=stop_after_attempt(3)):
    with attempt:
        # 需要重试的代码
        ...

最佳实践与风险提示

  • 明确重试的适用场景,避免对不可恢复的错误进行重试。
  • 合理设置重试次数和等待策略,防止资源浪费和系统压力过大。
  • 配合日志、告警等手段,及时发现和定位异常。
  • 避免无限重试或无等待重试,防止引发连锁故障。

总结

Tenacity 为 Python 提供了灵活、强大的重试能力,适用于多种同步与异步场景。合理配置重试策略,能有效提升程序的健壮性和容错性。更多细节请参见官方文档。