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.php.cn/link/85c19375f0c12c6793bf66b4e2666dc4") # 自动重试最多3次
自定义重试逻辑:捕获异常+手动控制
当需要更精细判断(如根据响应内容决定是否重试)、或集成日志、监控、熔断时,建议手动实现重试循环。
立即学习“Python免费学习笔记(深入)”;
- 用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(multiplier=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.php.cn/link/4d2fe2e8601f7a8018594d98f28706f2") print(result)
别忽略重试的副作用与边界
重试不是万能解药,滥用反而加剧问题。必须结合业务语义审慎设计:
- 幂等性优先:GET、HEAD、OPTIONS天然幂等;POST/PUT/DELETE需确认服务端是否支持幂等(如携带Idempotency-Key头)
- 超时要分层:连接超时(connect timeout)通常设1~3秒,读取超时(read timeout)视业务定(如API调用5~30秒),总耗时=重试次数×(连接+读取)
- 监控与告警:记录重试次数、最终失败原因(是网络问题?还是500频发?),便于定位真实瓶颈
- 降级兜底:重试全部失败后,应返回缓存数据、默认值或明确错误提示,而非让上游持续等待










