HTTP请求重试需合理控制时机、次数、条件和退避策略：优先用requests+urllib3内置Retry（支持状态码、异常类型、指数退避、方法限制）；复杂场景手动实现（捕获异常、业务判断、随机抖动）；异步推荐aiohttp+tenacity；须重视幂等性、分层超时、监控告警与降级兜底。
HTTP请求失败很常见，网络波动、服务端临时不可用、限流等都可能导致请求中断。Python中实现可靠的重试机制，关键不是“多试几次”，而是合理控制重试时机、次数、条件和退避策略。
用requests + urllib3内置重试最轻量
requests底层基于urllib3，其Retry类已封装完整重试逻辑，无需额外依赖。适合大多数同步请求场景。
- 指定要重试的HTTP状态码（如502、503、504、429）和连接异常类型（如ConnectTimeout、MaxRetryError）
- 设置backoff_factor启用指数退避：第n次重试前等待
backoff_factor × (2^(n-1))秒（例如factor=1时，等待1→2→4→8秒） - 限制最大重试次数（total）和针对特定方法的重试（如禁用POST重试避免重复提交）
示例：
from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry import requestssession = requests.Session() retry_strategy = Retry( total=3, status_forcelist=[429, 502, 503, 504], method_whitelist=["HEAD", "GET", "OPTIONS"], backoff_factor=1 ) adapter = HTTPAdapter(max_retries=retry_strategy) session.mount("http://", adapter) session.mount("https://", adapter)
response = session.get("https://www./link/85c19375f0c12c6793bf66b4e2666dc4") # 自动重试最多3次
自定义重试逻辑：捕获异常+手动控制
当需要更精细判断（如根据响应内容决定是否重试）、或集成日志、监控、熔断时，建议手动实现重试循环。
- 用try/except捕获requests.exceptions.RequestException及其子类（ConnectionError、Timeout、HTTPError等）
- 对response.status_code做业务级判断：比如401可刷新token后重试，400则直接放弃
- 使用time.sleep()配合随机抖动（jitter），避免重试请求扎堆冲击服务端
示例（带抖动的简单重试）：
import time import random import requestsdef fetch_with_retry(url, max_retries=3): for i in range(max_retries + 1): try: resp = requests.get(url, timeout=5) resp.raise_for_status() # 抛出4xx/5xx异常 return resp except requests.exceptions.RequestException as e: if i == max_retries: raise e
指数退避 + 最多0.5秒随机抖动
sleep_time = min(2 ** i + random.uniform(0, 0.5), 10) time.sleep(sleep_time) return None异步场景：aiohttp + tenacity组合更稳妥
在asyncio项目中，不能直接用time.sleep或requests。推荐aiohttp搭配重试库tenacity——它原生支持async/await，配置灵活。
- tenacity提供@retry装饰器，可按异常类型、返回值、耗时等条件触发重试
- 支持wait_exponential()（指数退避）、wait_random()（随机间隔）、stop_after_attempt()等策略组合
- 避免在协程中阻塞，所有等待通过await asyncio.sleep()完成
示例：
import asyncio import aiohttp from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type@retry( stop=stop_after_attempt(3), wait=wait_exponential(mu
ltiplier=1, min=1, max=10), retry=retry_if_exception_type((aiohttp.ClientError, asyncio.TimeoutError)) ) async def fetch_async(session, url): async with session.get(url, timeout=5) as response: if response.status >= 400: raise aiohttp.ClientResponseError( response.request_info, response.history, status=response.status ) return await response.text()
async def main(): async with aiohttp.ClientSession() as session: result = await fetch_async(session, "https://www./link/4d2fe2e8601f7a8018594d98f28706f2") print(result)
ltiplier=1, min=1, max=10),
retry=retry_if_exception_type((aiohttp.ClientError, asyncio.TimeoutError))
)
async def fetch_async(session, url):
async with session.get(url, timeout=5) as response:
if response.status >= 400:
raise aiohttp.ClientResponseError(
response.request_info, response.history, status=response.status
)
return await response.text()别忽略重试的副作用与边界
重试不是万能解药，滥用反而加剧问题。必须结合业务语义审慎设计：
- 幂等性优先：GET、HEAD、OPTIONS天然幂等；POST/PUT/DELETE需确认服务端是否支持幂等（如携带Idempotency-Key头）
- 超时要分层：连接超时（connect timeout）通常设1~3秒，读取超时（read timeout）视业务定（如API调用5~30秒），总耗时=重试次数×（连接+读取）
- 监控与告警：记录重试次数、最终失败原因（是网络问题？还是500频发？），便于定位真实瓶颈
- 降级兜底：重试全部失败后，应返回缓存数据、默认值或明确错误提示，而非让上游持续等待
