在 API 调用过程中，请求失败是不可避免的问题。理解错误码含义并构建智能重试机制，能显著提升系统稳定性。以下从错误码解析、重试策略设计、实战案例三个维度展开说明：
一、常见 API 错误码分类与处理逻辑
不同平台的错误码体系可能不同，但核心分类逻辑一致。以 HTTP 状态码为基础，结合平台自定义错误码，可分为以下几类：
- 客户端问题（需开发者修正）
错误码 典型场景 处理建议
400 参数格式错误（如日期格式 YYYY-MM-DD 写成 YYYY/MM/DD） 检查参数类型和格式，添加客户端校验
401 认证失败（token 过期、签名错误） 刷新 token 或重新生成签名
403 权限不足（未申请接口权限） 联系平台申请权限
404 请求路径不存在（如 URL 拼写错误） 核对文档中的 API 路径
429 限流（短时间内请求次数超限额） 降级处理（如缓存旧数据），后续实现智能重试 - 服务端问题（可重试或等待修复）
错误码 典型场景 处理建议
500 服务器内部错误（临时故障） 指数退避重试（如等待 1、2、4 秒后重试）
502 网关错误（服务器集群通信异常） 同上
503 服务不可用（服务器过载或维护） 同上，同时设置最大重试次数（如 5 次）
504 请求超时（服务器处理时间过长） 增加超时时间，或分段请求 - 业务逻辑错误（需人工干预）
错误码 典型场景 处理建议
400 业务规则不允许（如余额不足支付） 提示用户或触发其他业务流程
409 资源冲突（如重复创建同一订单） 检查业务唯一性约束
600+ 平台自定义错误（如订单状态异常） 参考平台文档处理特定错误
二、智能重试机制设计的 7 个关键要素
构建高效的重试机制需要平衡资源消耗与成功率，以下是核心设计要素： - 可重试错误码白名单
明确哪些错误需要重试（如 500、502、503），哪些需要直接失败（如 401、403）。示例配置：
python
运行
RETRYABLE_ERRORS = {
500, 502, 503, 504, # HTTP状态码
"SYSTEM_TIMEOUT", # 平台自定义错误码
"TEMPORARY_UNAVAILABLE"
} - 重试间隔策略
固定间隔：每次重试等待相同时间（如 3 秒），简单但不灵活。
指数退避：等待时间按指数增长（如 1、2、4、8 秒），适合服务端临时过载场景。
带抖动的指数退避：在指数退避基础上添加随机抖动（如 1±0.2 秒、2±0.4 秒），避免大量请求同时重试导致 “重试风暴”。 - 最大重试次数
防止无限重试消耗资源，通常设为 3-5 次。示例：
python
运行
MAX_RETRIES = 5 # 最多重试5次 - 超时设置
每次请求需设置合理超时时间（避免长时间等待无响应的请求）。例如：
python
运行
主请求超时时间（秒）
REQUEST_TIMEOUT = 10
重试总时间限制（秒）
TOTAL_RETRY_TIMEOUT = 60 - 重试上下文传递
记录重试次数、上次错误信息等，便于调试和审计。示例：
python
运行
{
"original_request": {"url": "https://api.example.com", "params": {...}},
"retry_count": 3,
"last_error": {"code": 503, "message": "Service Unavailable"}
} - 幂等性保障
对非幂等请求（如创建订单）避免重复提交。可通过以下方式实现：
生成唯一请求 ID（如 UUID），服务端根据 ID 去重。
先查询操作结果（如查询订单是否已创建），再决定是否重试。 - 熔断机制
当错误率持续高于阈值（如 30%）时，暂时停止重试，避免加重服务端负担。示例：
python
运行
使用熔断器（如Hystrix或自定义实现）
if error_rate > 0.3:
circuit_breaker.open() # 打开熔断器，暂停请求
raise CircuitBreakerError("Too many failures, circuit opened")
三、Python 实现重试机制的 3 种方案
方案 1：手动实现基础重试
python
运行
import requests
import time
def request_with_retry(url, params=None, max_retries=3, retry_delay=1):
retries = 0
while retries <= max_retries:
try:
response = requests.get(url, params=params, timeout=10)
if response.status_code in RETRYABLE_ERRORS:
raise Exception(f"Server error: {response.status_code}")
return response
except Exception as e:
retries += 1
if retries > max_retries:
raise # 超过最大重试次数，抛出异常
# 指数退避：每次等待时间翻倍
wait_time = retry_delay * (2 ** (retries - 1))
print(f"Request failed, retrying in {wait_time} seconds...")
time.sleep(wait_time)
方案 2：使用 tenacity 库（推荐）
tenacity是 Python 的重试库，支持多种重试策略，可通过装饰器简化代码：
python
运行
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
@retry(
stop=stop_after_attempt(5), # 最多重试5次
wait=wait_exponential(multiplier=1, min=2, max=10), # 指数退避，初始2秒，最大10秒
retry=retry_if_exception_type((requests.exceptions.ConnectionError,
requests.exceptions.Timeout)) # 特定异常重试
)
def call_api(url, params=None):
response = requests.get(url, params=params, timeout=10)
response.raise_for_status() # 非200状态码抛出异常
return response
方案 3：集成熔断器（应对服务雪崩）
使用pybreaker库实现熔断器模式：
python
运行
import pybreaker
import requests
初始化熔断器：最大失败3次，自动恢复时间60秒
circuit_breaker = pybreaker.CircuitBreaker(fail_max=3, reset_timeout=60)
@circuit_breaker
def call_api_with_circuit_breaker(url):
try:
response = requests.get(url, timeout=10)
if response.status_code in RETRYABLE_ERRORS:
raise Exception(f"Retriable error: {response.status_code}")
return response
except Exception as e:
raise # 触发熔断器计数
四、实战优化建议
监控与告警
统计重试成功率、平均重试次数等指标（如 Prometheus + Grafana）。
当特定错误码（如 403）频繁出现时触发告警，提示权限配置问题。
分环境配置
开发环境重试间隔短（如 1 秒），生产环境延长（如 3 秒），避免频繁重试影响服务端。
与缓存结合
若重试多次仍失败，返回缓存的旧数据（如 Redis 中缓存的 API 响应）。
日志增强
记录完整请求上下文（如请求 ID、重试次数、耗时），便于排查问题。
灰度测试
新接口上线时，先对少量请求开启重试机制，观察效果后再全量放开。
五、典型错误码处理案例
案例 1：处理 429 限流错误
python
运行
from tenacity import retry, wait_fixed, stop_after_attempt
@retry(
wait=wait_fixed(60), # 等待60秒（根据API限流周期调整）
stop=stop_after_attempt(3),
retry=lambda retry_state: retry_state.outcome.result().status_code == 429
)
def call_rate_limited_api(url):
response = requests.get(url)
return response
案例 2：处理 token 过期（401 错误）
python
运行
def refresh_token():
# 刷新token的逻辑
new_token = ...
return new_token
def call_api_with_auth(url):
token = get_current_token()
headers = {"Authorization": f"Bearer {token}"}
response = requests.get(url, headers=headers)
if response.status_code == 401:
# token过期，刷新后重试
new_token = refresh_token()
headers = {"Authorization": f"Bearer {new_token}"}
return requests.get(url, headers=headers)
return response
六、常见误区
盲目重试所有错误
对 400、403 等不可重试错误重试会浪费资源，应直接返回失败。
固定间隔重试
多个客户端同时固定间隔重试可能导致服务端压力骤增，推荐使用带抖动的指数退避。
忽略幂等性
对非幂等操作（如扣款）重试可能导致重复操作，需业务层保障。
无限重试
必须设置最大重试次数和总超时时间，防止资源耗尽。
通过合理解析错误码并构建智能重试机制，可将 API 调用成功率从 90% 提升至 99% 以上，显著增强系统稳定性。建议根据业务场景选择合适的重试方案，并持续优化参数配置
