在 Python 的 requests 库中实现重试机制可以通过以下几种方式完成，具体选择取决于你的需求（是否需要第三方库、是否需要自定义重试条件等）。
（图片来源网络，侵删）
方法 1：使用 requests.Session + 循环（简单手动重试）
适用于需要轻量级重试的场景,不依赖额外库。
import requests
from requests.exceptions import RequestException
def retry_request(url, max_retries=3, delay=1, **kwargs):
retries = 0
while retries < max_retries:
try:
response = requests.get(url, **kwargs)
response.raise_for_status() # 检查HTTP错误
return response
except RequestException as e:
retries += 1
print(f"请求失败，第 {retries} 次重试... 错误: {e}")
if retries < max_retries:
import time
time.sleep(delay) # 延迟重试
raise Exception(f"请求失败，已重试 {max_retries} 次")
# 使用示例
try:
resp = retry_request("https://example.com", max_retries=3)
print(resp.text)
except Exception as e:
print(e)
特点：
- 简单直接,无需额外依赖。
- 需要手动处理延迟和异常。
方法 2：使用 requests.adapters + HTTPAdapter（推荐）
通过 requests.Session 的 HTTPAdapter 配置重试，适合需要更精细控制的场景。
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
def create_session_with_retry(
max_retries=3,
backoff_factor=0.3,
status_forcelist=(500, 502, 503, 504),
session=None
):
session = session or requests.Session()
retry = Retry(
total=max_retries,
read=max_retries,
connect=max_retries,
backoff_factor=backoff_factor,
status_forcelist=status_forcelist,
)
adapter = HTTPAdapter(max_retries=retry)
session.mount("http://", adapter)
session.mount("https://", adapter)
return session
# 使用示例
session = create_session_with_retry(max_retries=3)
try:
resp = session.get("https://example.com")
resp.raise_for_status()
print(resp.text)
except requests.exceptions.RequestException as e:
print(f"请求失败: {e}")
特点：
（图片来源网络，侵删）
- 基于
urllib3的Retry机制，功能更强大。 - 支持指数退避（
backoff_factor）和指定状态码重试。 - 可复用
Session对象。
方法 3：使用第三方库 tenacity（灵活的重试装饰器）
如果需要更复杂的重试逻辑（如条件重试、多种异常处理），推荐 tenacity。
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
import requests
@retry(
stop=stop_after_attempt(3), # 最多重试3次
wait=wait_exponential(multiplier=1, min=1, max=10), # 指数退避
retry=retry_if_exception_type(requests.exceptions.RequestException), # 仅对特定异常重试
)
def fetch_with_retry(url, **kwargs):
response = requests.get(url, **kwargs)
response.raise_for_status()
return response
# 使用示例
try:
resp = fetch_with_retry("https://example.com")
print(resp.text)
except Exception as e:
print(f"最终失败: {e}")
特点：
- 支持灵活的重试条件（如异常类型、返回值等）。
- 可配置退避策略（固定延迟、指数退避等）。
- 代码更清晰,适合复杂场景。
方法 4：使用 requests-toolbelt（轻量级扩展）
requests-toolbelt 提供了 requests 的扩展功能，包括重试。
from requests_toolbelt.adapters import host_header_ssl_adapter
session = requests.Session()
session.mount('https://', host_header_ssl_adapter.HostHeaderSSLAdapter())
# 结合自定义重试逻辑（需手动实现）
#（此库更侧重于其他功能，重试需结合方法1或2）
总结对比
| 方法 | 优点 | 缺点 | 适用场景 |
|---|---|---|---|
| 手动循环 | 简单，无依赖 | 需要手动处理延迟和异常 | 简单脚本，快速实现 |
HTTPAdapter |
官方推荐，支持指数退避 | 需要理解 urllib3 的 Retry |
生产环境，需要稳定重试 |
tenacity |
灵活，支持复杂条件 | 需要额外安装库 | 需要精细控制重试逻辑的场景 |
requests-toolbelt |
轻量级扩展 | 重试功能较弱 | 其他功能需求（如HostHeader） |
推荐选择：
（图片来源网络，侵删）
- 一般场景：使用 方法2（
HTTPAdapter）。 - 需要复杂逻辑：使用 方法3（
tenacity）。 - 快速原型：使用 方法1（手动循环）。
注意事项
- 避免无限重试：始终设置最大重试次数（
max_retries）。 - 延迟策略：网络请求应配合指数退避（
backoff_factor），避免短时间内高频重试。 - 异常处理：区分可重试异常（如超时、5xx错误）和不可重试异常（如404、认证失败）。
- 资源释放：如果使用
Session，确保在代码中正确关闭（session.close()）。
