在Python中，实现重试机制最推荐的方式是使用tenacity或retry这类成熟库，它们内置了指数退避和异常过滤，能适应大多数场景；如果你需要更轻量或更定制化的控制，自己写一个装饰器也不复杂，核心在于：根据失败类型决定是否重试，控制好重试次数与间隔，避免无脑重试导致系统雪崩。
为什么你的代码需要重试机制
网络请求的不可靠性
无论是爬虫抓取数据，还是调用第三方API，网络波动是常态，很多失败是暂时的，比如DNS解析超时、连接被重置、服务器返回503，业内专家指出，在分布式系统中，临时性故障占比相当高,而重试机制正是应对这种不确定性的标准手段。
临时性故障与永久性故障
重试机制只适合处理临时性故障，例如HTTP 503（服务不可用）可以考虑重试，但401（未授权）或400（参数错误）重试多少次都没用，重试的前提是精准过滤异常，多数情况下，你需要在重试逻辑中区分异常类型,避免对永久性故障浪费资源。
重试适用的典型场景
- 爬虫请求：遇到反爬、IP临时被封、超时。
- 接口调用：第三方服务不稳定,返回5xx错误。
- 数据库连接：连接池短暂耗尽,重连后恢复。
- 文件写入：磁盘满或权限问题（需要区分，大部分是永久性）。
python重试机制最佳实践
使用tenacity库
tenacity是目前最流行的重试库，它支持装饰器风格,一行代码就能给函数加上重试功能。
from tenacity import retry, stop_after_attempt, wait_exponential
@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10))
def fetch_url(url):
# 网络请求
pass
核心参数：
stop：停止条件,如最大重试次数。wait：等待策略，如固定的、指数退避、随机抖动。retry：异常判断，默认所有异常都会重试,可以自定义只重试特定异常。
使用retry库
retry库更轻量，但功能相对简单,适合快速集成。
from retry import retry
@retry(tries=3, delay=2, backoff=2, max_delay=10)
def fetch_url(url):
pass
tries是重试次数，delay是初始延迟，backoff是退避因子,每次重试后延迟乘以该因子。
自定义装饰器（灵活控制）
如果不想引入额外依赖，或者需要特殊逻辑，可以自己写一个装饰器，核心逻辑包括：循环尝试、捕获异常、判断是否重试、延迟等待。
import time
from functools import wraps
def retry(max_retries=3, delay=1, exceptions=(Exception,)):
def decorator(func):
@wraps(func)
def wrapper(args, kwargs):
for attempt in range(max_retries):
try:
return func(args, kwargs)
except exceptions as e:
if attempt == max_retries - 1:
raise
time.sleep(delay (2 attempt)) # 指数退避
return None
return wrapper
return decorator
这个例子实现了指数退避，但缺乏抖动和细粒度异常过滤,不过对于个人项目已经够用。
退避策略详解
重试间隔直接影响系统压力。指数退避是最常用的策略：每次重试后等待时间翻倍，再加一点随机抖动,避免大量请求同时重试导致雪崩。
- 固定间隔：简单但容易引发峰值。
- 指数退避：初始间隔较小，后续指数增长,适合大多数场景。
- 指数退避+抖动：在指数退避基础上增加随机范围，进一步分散重试时间,推荐用于高并发环境。
在tenacity中，可以通过wait_exponential和wait_random组合实现抖动。
python重试库对比：tenacity vs retry
功能对比
| 特性 | tenacity | retry |
|---|---|---|
| 装饰器支持 | 完整 | 完整 |
| 指数退避 | 内置多种策略 | 通过backoff参数 |
| 随机抖动 | 支持 | 无 |
| 异步支持 | 支持asyncio | 不支持 |
| 自定义异常过滤 | 丰富 | 仅支持tuple |
| 回调函数 | 支持（before/after） | 无 |
| 安装大小 | 约50KB | 约10KB |
| 适用场景 | 复杂生产环境 | 简单脚本、快速迭代 |
性能对比
在多数情况下，tenacity的延迟处理更精细，但retry的代码量更小，如果你的场景是异步爬虫，tenacity是首选；如果只是给一个函数加简单重试,retry更直接。
如何选择
- 需要异步重试：用tenacity。
- 需要复杂退避策略（如抖动）：用tenacity。
- 追求最小依赖：写自定义装饰器或retry。
- 需要重试前后回调（如日志记录）：用tenacity。
python重试场景实战
爬虫重试：避免频繁被封
在爬虫中，重试次数不宜过多，且要配合随机延时和更换代理，行业共识认为，重试3次，每次间隔指数退避，同时轮换User-Agent,能有效降低被封概率。
@retry(stop=stop_after_attempt(3), wait=wait_random(min=1, max=3))
def scrape(url, headers):
resp = requests.get(url, headers=headers, timeout=5)
if resp.status_code == 429: # 限流
raise Exception("Too Many Requests")
return resp
接口调用重试：考虑超时
调用第三方API时，除了网络异常，还要关注HTTP状态码，对5xx重试，对4xx不重试，同时设置总超时,避免重试时间过长。
@retry(stop=stop_after_attempt(3), retry=retry_if_exception(should_retry))
def call_api(payload):
resp = requests.post(API_URL, json=payload, timeout=3)
if resp.status_code >= 500:
raise Exception("Server error")
return resp.json()
异步任务重试
配合asyncio,tenacity能无缝支持异步函数。
@retry(stop=stop_after_attempt(3), wait=wait_exponential())
async def async_fetch(url):
async with aiohttp.ClientSession() as session:
async with session.get(url) as resp:
return await resp.text()
python重试常见问题
重试机制如何设置超时？
重试的超时包括单次请求超时和总超时，单次超时在请求级别设置（如timeout=5），总超时则需要计算重试次数和间隔，tenacity没有直接提供总超时，但可以结合stop_after_delay实现，例如stop=stop_after_delay(30)，表示最多重试30秒，对于超时敏感的接口，建议设置总超时并配合指数退避,避免单次重试累积时间过长。
重试太多次影响性能怎么办？
控制重试次数和间隔，并设置最大延迟，使用指数退避+抖动可以避免同时重试导致服务雪崩，可以引入断路器模式，在连续失败达到阈值后直接放弃，等待一段时间再恢复，在tenacity中，可以用stop_after_attempt限制次数，用wait_exponential设置最大延迟,确保重试不会无限进行。
重试还是不重试？如何判断？
判断标准是失败类型，对于幂等操作（如GET请求、写入后无副作用），可以放心重试，对于非幂等操作（如支付、下单），必须谨慎，最好增加幂等键，确保重试不会重复执行，对于临时性故障（网络超时、503），重试；对于永久性故障（参数错误、无权限），不重试，在代码中，通过异常过滤只对特定异常重试,是避免误判的关键。
重试机制是提升代码健壮性的常用手段，但也要防止滥用，选择合适的重试库、退避策略和异常过滤，能让你的程序在复杂网络环境中依然稳定。少即是多，重试几次就够了,别让代码变成无休止的尝试。
首发原创文章，作者：王坚，如若转载，请注明出处：https://test.idctop.com/article/507954.html
