Python 接口调用不稳定?一文掌握超时控制、自动重试与日志记录
一、为什么接口调用不能只写一行代码?
在很多项目中,调用 HTTP 接口可能只需要下面几行代码:
import requests
response = requests.get("https://example.com/api")
print(response.json())
这种写法简单直观,但在实际运行中仍有几个问题:
- 请求可能无限等待。 如果没有设置超时,程序可能长时间阻塞。
- 临时故障可能导致任务失败。 网络波动、服务端过载等问题不一定意味着请求永远无法成功。
- HTTP 错误容易被忽略。 收到 HTTP 500 响应并不等于请求成功。
- 异常缺少上下文。 如果没有记录请求地址、耗时和错误类型,排查问题就会比较困难。
因此,一个更健壮的请求客户端通常需要具备超时控制、合理的自动重试、明确的错误处理和必要的日志记录。
二、设置合理的请求超时
requests 支持通过 timeout 参数控制连接等待和读取响应的时间。
import requests
response = requests.get(
"https://example.com/api",
timeout=(3, 10)
)
response.raise_for_status()
data = response.json()
print(data)
这里的 (3, 10) 分别表示连接超时和读取超时,单位均为秒。
需要注意,读取超时主要限制等待服务器发送数据的间隔,并不等同于整个请求从开始到结束的绝对总时长。
实践建议:
- 根据接口的实际响应时间设置超时。
- 对响应时间较长的业务接口进行单独配置。
- 不要因为担心请求失败,就无限放大超时时间。
- 对批量任务设置整体执行时间或任务级截止时间,避免单个请求拖累整个任务。
三、为临时故障增加自动重试
对于部分临时性故障,可以在请求失败后进行有限次数的重试,并通过逐渐增加等待时间来降低连续请求对服务端造成的压力。
安装依赖:
pip install requests
下面使用 HTTPAdapter 和 Retry 配置重试策略:
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
def create_session():
retry_strategy = Retry(
total=3,
connect=3,
read=2,
status=3,
backoff_factor=0.5,
status_forcelist=[429, 500, 502, 503, 504],
allowed_methods=frozenset(
["GET", "HEAD", "OPTIONS"]
),
respect_retry_after_header=True,
)
adapter = HTTPAdapter(
max_retries=retry_strategy
)
session = requests.Session()
session.mount("https://", adapter)
session.mount("http://", adapter)
return session
if __name__ == "__main__":
session = create_session()
try:
response = session.get(
"https://example.com/api",
timeout=(3, 10),
)
response.raise_for_status()
print(response.json())
except requests.RequestException as exc:
print(f"接口请求失败:{exc}")
finally:
session.close()
这段代码实现了以下功能:
- 使用
Session复用请求配置。 - 对部分连接异常和指定 HTTP 状态码配置有限次数的重试。
- 通过
backoff_factor设置退避等待策略。 - 在服务端提供
Retry-After响应头时,允许重试机制参考该提示。 - 在请求最终失败时,统一捕获
requests.RequestException。
这里的重试次数是配置上限,不应简单理解为所有异常都会严格重试相同次数。具体行为还会受到异常类型、HTTP 方法和 urllib3 版本等因素影响。
四、自动重试有哪些注意事项?
自动重试不是越多越好,尤其是在涉及数据库写入、订单创建、支付和其他状态变更的业务中。
1. 优先考虑幂等性
幂等操作是指在满足相应业务条件时,对同一操作重复执行不会导致最终状态不断变化。
例如,查询资源的 GET 请求通常适合进行有限重试;而创建订单的 POST 请求如果已经在服务端执行成功,只是响应丢失,那么再次发送可能导致重复创建。
对于写操作,应结合业务设计使用幂等键、请求唯一标识或服务端去重机制,而不是简单地为所有 POST 请求开启自动重试。
2. 不要把所有错误都当作临时故障
常见错误可以按以下思路处理:
| 错误类型 | 建议处理方式 |
|---|---|
| 连接超时 | 评估网络问题后,有限重试 |
| HTTP 429 | 遵循服务端限流提示,控制请求频率 |
| HTTP 500、502、503、504 | 根据业务情况有限重试 |
| HTTP 400 | 检查请求参数,通常不应直接重试 |
| HTTP 401、403 | 检查认证信息与权限 |
| JSON 解析失败 | 检查响应格式,不应默认重新发送请求 |
具体处理方式仍应以接口文档和业务语义为准。
3. 防止重试放大故障
如果大量客户端同时遇到服务端故障,并且立即重复请求,可能进一步增加服务端压力。
因此,生产环境中通常需要综合考虑:
- 最大重试次数。
- 退避等待时间。
- 请求并发量。
- 整体请求时限。
- 服务端限流规则。
- 必要时使用随机抖动,避免大量请求同时重试。
五、增加日志记录,让故障更容易定位
只在终端打印错误,往往不足以支持长期运行的任务排查。可以使用 Python 内置的 logging 模块记录请求耗时和失败原因。
import logging
import time
import requests
logging.basicConfig(
level=logging.INFO,
format=(
"%(asctime)s "
"%(levelname)s "
"%(message)s"
),
)
def get_json(session, url):
start = time.monotonic()
try:
response = session.get(
url,
timeout=(3, 10),
)
response.raise_for_status()
data = response.json()
elapsed = time.monotonic() - start
logging.info(
"请求成功,耗时 %.3f 秒",
elapsed,
)
return data
except requests.RequestException:
elapsed = time.monotonic() - start
logging.exception(
"请求失败,耗时 %.3f 秒",
elapsed,
)
raise
except ValueError:
logging.exception("响应 JSON 解析失败")
raise
该函数可以与前面的 create_session() 配合使用。需要注意,ValueError 也可能来自其他数据转换过程,因此在更复杂的业务中,可以进一步缩小异常捕获范围。
在真实项目中,还可以记录请求标识、业务任务 ID、目标服务名称和最终状态码,帮助定位问题。
安全提醒: 不要在日志中输出密码、访问令牌、完整的认证请求头或其他敏感数据。对用户信息和业务数据也应遵循最小化记录原则。
六、如何验证重试机制是否生效?
建议通过受控测试验证行为,而不是直接在生产环境中制造故障。
可以重点检查以下场景:
- 正常响应: 请求成功后,程序能否正常解析 JSON。
- 连接失败: 请求最终失败时,是否能够退出并报告异常。
- 指定错误状态码: 对配置中的临时错误,是否按预期重试。
- 非重试状态码: 对不应重试的请求,是否能够尽快返回错误。
- 耗时与日志: 日志是否包含足够的排查信息,同时没有泄露敏感数据。
- 依赖版本: 在项目实际使用的
requests和urllib3版本下,确认重试参数与行为符合预期。
可以通过本地测试服务器或专门的模拟接口返回不同状态码,验证请求次数、等待行为和最终结果。不要在未授权的第三方服务上开展压力测试。
七、总结
构建可靠的 HTTP 客户端,不只是给请求增加一个 try/except,而是需要从请求时限、故障分类、重试策略和可观测性几个方面综合设计。
本文的核心建议是:
- 为请求设置合理的连接和读取超时。
- 只对适合重试的请求及错误配置有限重试。
- 对涉及状态变更的操作,优先解决幂等性和重复执行问题。
- 通过结构化日志提高故障定位效率。
- 使用受控测试验证实际行为,并结合业务需求调整参数。
当这些基础机制设计合理后,Python 接口调用通常会更容易维护,也更便于集成到云上定时任务、数据处理脚本和后台服务中。
- 点赞
- 收藏
- 关注作者
评论(0)