超时、代理、重试这些问题我都遇到过
事情是这样的,上个月给客户做一个数据采集系统,需要调用一个第三方API获取商品价格数据
代码本地跑得好好的,一部署到服务器就各种玄学问题——超时、连接被拒绝、返回500错误,那几天快把我折磨疯了
今天把踩过的坑整理一下,都是真实经历,看完能少走弯路
第一个坑:超时设置了个寂寞
一开始代码很简单,就类似这样:
import requests
defget_price(product_id):
url = f"https://api.example.com/product/{product_id}"
response = requests.get(url)
return response.json()
本地测试没问题啊,调用第三方API响应还挺快的
结果部署到服务器,第一次请求就卡住了,整整等了5分钟才报错:
requests.exceptions.ReadTimeout: HTTPSConnectionPool(host='api.example.com', port=443): Read timed out. (read timeout=None)
我一看这报错就懵了,read timeout=None是什么鬼
我明明没设置超时,但也不能让我无限等下去啊
后来查了文档才发现,requests默认没有超时设置,如果服务器不响应,它真的会一直等下去
生产环境网络环境复杂,服务器可能各种原因不响应,这个坑是真的坑
修复很简单,加上timeout参数:
response = requests.get(url, timeout=10)
但这里有个细节我后来才注意到——timeout参数实际上包含两个值:(connect_timeout, read_timeout)
如果只传一个数字,那它会同时作为连接超时和读取超时
正确做法应该是:
# 连接超时5秒,读取超时30秒
response = requests.get(url, timeout=(5, 30))
或者更保守一点:
# 连接和读取都是10秒
response = requests.get(url, timeout=10)
第二个坑:代理设置了但没用
加上超时后,接口能返回了,但是响应很慢,而且经常失败
报错是这样的:
requests.exceptions.proxyError: Tunnel connection failed: 403 Forbidden
我寻思这不对啊,本地能跑,服务器上就不行了
后来问了运维才知道,服务器出网需要走代理
好,我加代理:
proxies = {
"http": "http://proxy.company.com:8080",
"https": "http://proxy.company.com:8080"
}
response = requests.get(url, timeout=10, proxies=proxies)
结果还是报403,换了各种代理格式都不行
折腾了一下午,后来发现是代理认证的问题——我们公司的代理需要用户名密码
正确写法应该是:
proxies = {
"http": "http://user:password@proxy.company.com:8080",
"https": "http://user:password@proxy.company.com:8080"
}
但这样写用户名密码暴露在代码里,很不安全
后来我用了环境变量:
import os
proxies = {
"http": os.environ.get("HTTP_PROXY"),
"https": os.environ.get("HTTPS_PROXY")
}
服务器上设置环境变量就行:
export HTTP_PROXY="http://user:password@proxy.company.com:8080"
export HTTPS_PROXY="http://user:password@proxy.company.com:8080"
这个坑踩得比较冤,requests的代理文档写得不太明显,没说清楚认证的写法
第三个坑:第三方API不稳定,请求总是失败
加上代理后,请求是能发出去了,但架不住第三方API不稳定啊
有时候返回500,有时候返回502,有时候干脆超时
客户那边可不管你这些借口,采集数据对不上就找你
没办法,只能加重试机制
一开始我手写了个简单的重试:
defget_price_with_retry(product_id, max_retries=3):
url = f"https://api.example.com/product/{product_id}"
for i inrange(max_retries):
try:
response = requests.get(url, timeout=(5, 30), proxies=proxies)
if response.status_code == 200:
return response.json()
except Exception as e:
print(f"第{i+1}次请求失败: {e}")
if i == max_retries - 1:
raise
returnNone
能用,但是太土了
每次请求都要写这一堆代码,重复代码一大堆
后来我发现了requests.adapters这个好东西,可以给Session配置自动重试:
from requests.adapters import HTTPAdapter
from requests.packages.urllib3.util.retry import Retry
import requests
defcreate_session():
session = requests.Session()
# 配置重试策略
retry_strategy = Retry(
total=3, # 最多重试3次
backoff_factor=1, # 重试间隔:1s, 2s, 4s
status_forcelist=[500, 502, 503, 504], # 这些状态码才重试
allowed_methods=["GET", "POST"]
)
adapter = HTTPAdapter(max_retries=retry_strategy)
session.mount("http://", adapter)
session.mount("https://", adapter)
return session
# 使用
session = create_session()
response = session.get(url, timeout=(5, 30), proxies=proxies)
这个方案好太多了,重试逻辑封装在Session级别,代码简洁,而且backoff_factor参数会自动计算重试间隔——第一次等1秒,第二次等2秒,第三次等4秒,避免对服务器造成压力
第四个坑:SSL证书验证失败
重试加上了,本地测试完美,结果服务器上报错:
requests.exceptions.SSLError: HTTPSConnectionPool(host='api.example.com', port=443):
certificate verify failed: unable to get local issuer certificate
又是本地没问题,服务器有问题
我估计是服务器的CA证书库没更新,或者第三方API的证书有问题
紧急修复,加verify=False:
response = session.get(url, timeout=(5, 30), proxies=proxies, verify=False)
但是这样有个问题——禁用SSL验证有安全风险,生产环境不建议这么做
更稳妥的做法是指定CA证书文件:
response = session.get(url, timeout=(5, 30), proxies=proxies,
verify="/path/to/ca-bundle.crt")
或者让运维更新服务器上的CA证书:
# Ubuntu/Debian
sudo apt-get update && sudo apt-get install ca-certificates
# CentOS/RHEL
sudo yum update ca-certificates
我最后用的是verify=False,因为那个第三方API确实证书有问题,临时方案先跑起来再说
第五个坑:Session和Cookie的恩怨
系统跑起来了,但有个奇怪的问题——登录接口调用成功,后续请求却说我没登录
代码是这样的:
# 登录
login_url = "https://api.example.com/login"
requests.post(login_url, data={"username": "xxx", "password": "yyy"})
# 获取商品列表
list_url = "https://api.example.com/products"
response = requests.get(list_url) # 这里说我没登录
问题在于,requests的每次请求都是独立的Session
登录成功后,服务端在Cookie里写入了会话信息,但第二个请求是新建的连接,根本没有Cookie
解决方案很简单,用Session对象:
session = requests.Session()
# 登录(Cookie会自动保存)
session.post(login_url, data={"username": "xxx", "password": "yyy"})
# 后续请求(带着Cookie)
response = session.get(list_url)
其实我上面已经用了Session配置重试,正好把Cookie问题一起解决了
有时候踩坑踩着踩着,就顺带把其他问题解决了
最终代码
整合一下,最终的代码大概是这样的:
import os
import requests
from requests.adapters import HTTPAdapter
from requests.packages.urllib3.util.retry import Retry
defcreate_session():
"""创建一个带有重试机制的Session"""
session = requests.Session()
# 重试策略
retry_strategy = Retry(
total=3,
backoff_factor=1,
status_forcelist=[500, 502, 503, 504],
allowed_methods=["GET", "POST"]
)
adapter = HTTPAdapter(max_retries=retry_strategy)
session.mount("http://", adapter)
session.mount("https://", adapter)
return session
defget_price(product_id):
"""获取商品价格"""
session = create_session()
# 代理配置
proxies = {
"http": os.environ.get("HTTP_PROXY"),
"https": os.environ.get("HTTPS_PROXY")
}
# 超时配置:(连接超时, 读取超时)
timeout = (10, 30)
url = f"https://api.example.com/product/{product_id}"
response = session.get(
url,
timeout=timeout,
proxies=proxies,
verify=False# 视情况而定
)
response.raise_for_status()
return response.json()
if __name__ == "__main__":
price = get_price("12345")
print(price)
这段代码在生产环境跑了一个月,目前还算稳定
总结一下这些坑
- 3. 重试机制用
requests.adapters,比手写优雅很多 - 4. SSL证书问题生产环境常遇到,知道怎么临时绕过
- 5. 保持会话用Session,别每次请求都new一个
requests这个库入门很简单,但要用好确实有不少细节
踩坑的过程也是学习的过程,这些问题现在看来都是基础,但当时确实被折磨得不轻
希望我的这些经历能给你提个醒,少走弯路