Python 重试神器 tenacity 详解：让不稳定代码自动重试更优雅

一、tenacity 库概述

tenacity 是 Python 中一款轻量且强大的重试装饰器库，核心用于为函数、方法添加自动重试逻辑，解决网络请求、接口调用、IO 操作等不稳定场景的执行失败问题。

其原理是通过装饰器无侵入式包裹目标代码，捕获指定异常后，按配置的等待策略、重试次数、退避算法重新执行函数，直到调用成功或达到终止条件。该库使用 Apache 2.0 开源协议，优点是语法简洁、策略丰富、无侵入、支持异步，缺点是仅专注重试逻辑，不提供故障熔断、降级能力。

二、tenacity 安装方法

tenacity 支持 Python 3.6 及以上版本，安装方式简单，直接使用 pip 命令即可完成安装：

pip install tenacity

若需要使用异步重试相关功能，无需额外安装依赖，tenacity 原生支持 asyncio 异步环境。

安装完成后，在 Python 脚本中直接导入即可使用，基础导入语句如下：

# 基础重试装饰器
from tenacity import retry
# 常用重试条件：指定异常类型重试
from tenacity import retry_if_exception_type
# 重试停止条件：限制重试次数
from tenacity import stop_after_attempt
# 重试等待条件：固定间隔等待
from tenacity import wait_fixed

三、tenacity 基础使用方式

3.1 最简重试示例（无任何配置）

tenacity 最基础的用法是直接给函数添加 @retry 装饰器，不配置任何参数，此时默认行为是：只要函数抛出任意异常，就无限重试，直到函数执行成功。

示例代码：

from tenacity import retry

# 计数器，记录函数执行次数
count = 0

@retry
def unstable_func():
    global count
    count += 1
    print(f"函数第 {count} 次执行")
    # 主动抛出异常，模拟执行失败
    raise Exception("执行失败，触发重试")

if __name__ == "__main__":
    try:
        unstable_func()
    except Exception as e:
        print(f"最终执行失败：{e}")

代码说明：

1. 定义一个不稳定函数 unstable_func，每次执行都会抛出异常；
2. 添加 @retry 装饰器后，函数会无限次重试执行；
3. 因为函数始终抛出异常，所以会一直循环执行，直到手动终止程序。

这种无配置方式适合临时调试，实际开发中必须限制重试次数，避免无限循环占用资源。

3.2 限制重试次数

通过 stop_after_attempt(n) 可以指定最多重试 n 次，注意：总执行次数 = 1 次初始执行 + n 次重试。

示例代码：

from tenacity import retry, stop_after_attempt

count = 0

# 最多重试 3 次，总执行 4 次
@retry(stop=stop_after_attempt(3))
def unstable_func():
    global count
    count += 1
    print(f"函数第 {count} 次执行")
    raise Exception("接口请求超时")

if __name__ == "__main__":
    try:
        unstable_func()
    except Exception as e:
        print(f"重试 3 次后仍失败：{e}")

代码说明：

1. stop=stop_after_attempt(3) 表示最多重试 3 次；
2. 函数会执行 1 次初始调用 + 3 次重试，共 4 次；
3. 4 次执行都失败后，不再重试，直接抛出原始异常。

3.3 设置重试等待时间

实际场景中，重试不能无间隔执行，否则会给服务器造成巨大压力，tenacity 提供多种等待策略，最常用的是固定等待时间 wait_fixed(秒数)。

示例代码：

from tenacity import retry, stop_after_attempt, wait_fixed
import time

count = 0

# 重试 3 次，每次重试前等待 2 秒
@retry(
    stop=stop_after_attempt(3),
    wait=wait_fixed(2)
)
def request_api():
    global count
    count += 1
    current_time = time.strftime("%H:%M:%S")
    print(f"【{current_time}】第 {count} 次请求接口")
    raise ConnectionError("网络连接失败")

if __name__ == "__main__":
    try:
        request_api()
    except ConnectionError as e:
        print(f"最终请求失败：{e}")

代码说明：

1. wait=wait_fixed(2) 表示每次重试前等待 2 秒；
2. 执行日志会清晰看到每次请求间隔 2 秒，避免高频请求；
3. 适合网络波动、接口限流等需要短暂等待的场景。

3.4 仅针对指定异常重试

很多场景下，不是所有异常都需要重试，比如参数错误、权限不足等异常，重试也不会成功，此时可以通过 retry_if_exception_type 指定只捕获特定异常。

示例代码：

from tenacity import retry, stop_after_attempt, wait_fixed, retry_if_exception_type

count = 0

# 仅捕获 ConnectionError 重试，其他异常直接抛出
@retry(
    stop=stop_after_attempt(3),
    wait=wait_fixed(1),
    retry=retry_if_exception_type(ConnectionError)
)
def request_data():
    global count
    count += 1
    print(f"第 {count} 次请求数据")
    # 模拟网络异常（会重试）
    raise ConnectionError("网络断开")
    # 若抛出 ValueError（不会重试）
    # raise ValueError("参数错误")

if __name__ == "__main__":
    request_data()

代码说明：

1. retry=retry_if_exception_type(ConnectionError) 限定只有网络连接异常才重试；
2. 如果函数抛出非指定异常（如 ValueError、TypeError），装饰器不会触发重试；
3. 精准控制重试逻辑，避免无效重试。

3.5 指数退避重试（高级等待策略）

针对第三方接口、云服务等场景，推荐使用指数退避算法，即等待时间随重试次数指数增长，避免集中请求，wait_exponential 是 tenacity 内置的指数退避策略。

示例代码：

from tenacity import retry, stop_after_attempt, wait_exponential
import time

count = 0

# 指数退避：初始等待 1 秒，最大等待 10 秒
@retry(
    stop=stop_after_attempt(5),
    wait=wait_exponential(multiplier=1, min=1, max=10)
)
def upload_file():
    global count
    count += 1
    current_time = time.strftime("%H:%M:%S")
    print(f"【{current_time}】第 {count} 次上传文件")
    raise TimeoutError("上传超时")

if __name__ == "__main__":
    upload_file()

代码说明：

1. multiplier=1 表示基础倍数，min=1 最小等待 1 秒，max=10 最大等待 10 秒；
2. 等待时间依次为：1s → 2s → 4s → 8s → 10s（达到最大值后不再增长）；
3. 适合调用付费接口、公共 API 等需要友好访问的场景。

3.6 重试前后执行自定义逻辑

tenacity 支持在每次重试前、重试后、最终失败时执行自定义函数，方便打印日志、记录状态、发送告警。

通过 before、after、stop 回调函数实现：

from tenacity import retry, stop_after_attempt, wait_fixed, before_log, after_log
import logging

# 配置日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

count = 0

@retry(
    stop=stop_after_attempt(3),
    wait=wait_fixed(1),
    # 重试前打印日志
    before=before_log(logger, logging.INFO),
    # 重试后打印日志
    after=after_log(logger, logging.INFO)
)
def download_data():
    global count
    count += 1
    print(f"第 {count} 次下载数据")
    raise Exception("下载失败")

if __name__ == "__main__":
    download_data()

代码说明：

1. before_log 在每次重试执行函数前打印日志；
2. after_log 在每次重试执行函数后打印日志；
3. 可以自定义普通函数替换日志函数，实现发送钉钉/微信告警、写入数据库等操作。

3.7 异步函数重试

tenacity 原生支持 Python asyncio 异步函数，用法和同步函数完全一致，只需给异步函数添加装饰器即可。

示例代码：

import asyncio
from tenacity import retry, stop_after_attempt, wait_fixed

count = 0

# 异步函数重试配置
@retry(
    stop=stop_after_attempt(3),
    wait=wait_fixed(1)
)
async def async_request():
    global count
    count += 1
    print(f"第 {count} 次异步请求")
    raise Exception("异步请求失败")

async def main():
    await async_request()

if __name__ == "__main__":
    asyncio.run(main())

代码说明：

1. 异步函数直接添加 @retry 装饰器，无需修改其他逻辑；
2. 等待、重试次数、异常过滤等配置和同步函数通用；
3. 适合异步爬虫、异步接口、异步 IO 等场景。

四、实际开发综合案例

4.1 案例场景

模拟爬虫请求第三方接口：网络不稳定、接口偶尔超时，需要实现：

1. 最多重试 5 次；
2. 指数退避等待，避免高频请求；
3. 仅捕获网络异常、超时异常重试；
4. 每次重试打印日志；
5. 最终失败返回默认值。

4.2 完整代码实现

from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
import time
import requests

# 自定义重试异常类型：网络错误、超时错误
RETRY_EXCEPTIONS = (requests.exceptions.ConnectionError, requests.exceptions.Timeout)

def log_retry(retry_state):
    """自定义重试日志函数"""
    print(f"【重试】第 {retry_state.attempt_number} 次失败，等待后重试")

@retry(
    # 最多重试 5 次
    stop=stop_after_attempt(5),
    # 指数退避：1s, 2s, 4s, 8s, 10s
    wait=wait_exponential(multiplier=1, min=1, max=10),
    # 仅针对指定网络异常重试
    retry=retry_if_exception_type(RETRY_EXCEPTIONS),
    # 重试前执行日志函数
    before_sleep=log_retry
)
def crawl_api(url: str) -> dict:
    """
    爬取第三方接口数据
    :param url: 接口地址
    :return: 接口返回数据
    """
    print(f"\n开始请求接口：{url}")
    response = requests.get(url, timeout=3)
    response.raise_for_status()  # 非200状态码抛出异常
    return response.json()

if __name__ == "__main__":
    api_url = "https://httpstat.us/503"  # 模拟服务不可用接口
    try:
        data = crawl_api(api_url)
        print("请求成功：", data)
    except Exception as e:
        print(f"\n重试 5 次后最终失败，返回默认数据")
        # 最终失败返回默认值，保证程序不崩溃
        default_data = {"code": -1, "msg": "接口请求失败", "data": []}
        print("默认数据：", default_data)

代码说明：

1. 封装通用爬虫函数，对接不稳定第三方接口；
2. 组合多种 tenacity 策略，兼顾稳定性与友好性；
3. 最终失败捕获异常，返回默认数据，保证主程序正常运行；
4. 可直接用于生产环境的爬虫、接口调用、数据同步等场景。

4.3 案例扩展：文件读取重试

模拟读取本地文件时，文件被占用、读取失败的场景，自动重试读取：

from tenacity import retry, stop_after_attempt, wait_fixed
import os

@retry(
    stop=stop_after_attempt(3),
    wait=wait_fixed(1)
)
def read_file(file_path: str) -> str:
    """读取本地文件，失败自动重试"""
    if not os.path.exists(file_path):
        raise FileNotFoundError("文件不存在")
    with open(file_path, "r", encoding="utf-8") as f:
        return f.read()

if __name__ == "__main__":
    try:
        content = read_file("data.txt")
        print("文件内容：", content)
    except Exception as e:
        print("文件读取失败：", e)

该示例适用于本地文件读写、日志读取、配置文件加载等 IO 场景。

相关资源

• Pypi地址：https://pypi.org/project/tenacity
• Github地址：https://github.com/jd/tenacity
• 官方文档地址：https://tenacity.readthedocs.io/

关注我，每天分享一个实用的Python自动化工具。