在电商物流追踪、金融数据监控等场景中，API请求的稳定性直接决定系统可靠性。当顺丰API因网络抖动返回503错误，或因跨地域调用出现10秒延迟时，如何确保程序不崩溃且数据不丢失？本文通过真实案例拆解，用Requests库实现"防抖动+抗异常"的健壮请求方案。
一、血泪教训：那些年踩过的API坑
某跨境电商系统在"黑色星期五"大促期间突发故障：调用顺丰国际件接口时，30%的请求因超时失败，导致2000+包裹状态同步延迟。事后分析发现三大元凶：
固定超时陷阱：设置timeout=5导致所有跨洋请求必然超时（实际平均响应时间8秒）
暴力重试雪崩：简单for循环重试5次，瞬间产生10倍请求量击垮顺丰网关
代理池污染：使用失效代理IP发起请求，触发顺丰反爬机制封禁整个IP段
这些场景揭示核心问题：API请求需要"有智慧的等待"和"有策略的坚持"。
二、超时配置：给请求装上"安全阀"
- 连接超时 vs 读取超时
import requests
try:
# 连接超时3秒（TCP握手阶段）
# 读取超时10秒（服务器处理阶段）
response = requests.get(
'https://api.sf-express.com/track',
params={'trackingNumber': 'SF123456789'},
timeout=(3, 10) # 元组形式分别设置
)
print(response.json())
except requests.exceptions.ConnectTimeout:
print("连接服务器失败，请检查网络")
except requests.exceptions.ReadTimeout:
print("服务器处理超时，请稍后重试")
关键决策点：
国内API调用：timeout=(2, 5)（连接2秒，读取5秒）
跨境API调用：timeout=(5, 15)（考虑国际链路延迟）
文件上传场景：需增加write_timeout参数（需httpx等库支持）
- 动态超时策略
某物流监控系统采用分级超时机制：
def get_dynamic_timeout(retry_count):
base_timeout = 3 # 基础超时
if retry_count > 0:
return min(base_timeout (2 * retry_count), 30) # 指数退避，最大30秒
return base_timeout
使用示例
for i in range(3):
try:
timeout = get_dynamic_timeout(i)
response = requests.get(url, timeout=timeout)
break
except Exception as e:
print(f"第{i+1}次尝试失败，超时时间调整为{timeout}秒")
三、重试机制：让请求学会"坚持"
- 指数退避重试（推荐方案）
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
def create_retry_session(retries=3, backoff_factor=1, status_forcelist=(500, 502, 503, 504)):
session = requests.Session()
retry = Retry(
total=retries,
read=True, # 允许读取超时重试
connect=True, # 允许连接超时重试
backoff_factor=backoff_factor,
status_forcelist=status_forcelist,
allowed_methods=["GET", "POST"] # 支持POST请求重试
)
adapter = HTTPAdapter(max_retries=retry)
session.mount("http://", adapter)
session.mount("https://", adapter)
return session
使用示例
session = create_retry_session()
response = session.get('https://api.sf-express.com/track', params={'trackingNumber': 'SF123456789'})
参数深解：
backoff_factor=1：第1次重试等待1秒，第2次2秒，第3次4秒
status_forcelist：仅对5xx服务器错误和429限流错误重试
allowed_methods：默认不重试POST请求，需显式声明
- 熔断机制实现（避免雪崩）
from collections import deque
import time
class CircuitBreaker:
def init(self, max_failures=3, reset_timeout=60):
self.failures = deque(maxlen=max_failures)
self.reset_timeout = reset_timeout
def is_open(self):
if len(self.failures) < self.failures.maxlen:
return False
# 如果最近max_failures次请求都失败，且最后一次失败在reset_timeout秒内
return (time.time() - self.failures[-1]) < self.reset_timeout
def record_failure(self):
self.failures.append(time.time())
def record_success(self):
self.failures.clear()
结合重试使用
breaker = CircuitBreaker(max_failures=3, reset_timeout=30)
def safe_request():
if breaker.is_open():
raise Exception("Service unavailable, circuit breaker open")
try:
response = create_retry_session().get(url)
if response.status_code == 200:
breaker.record_success()
return response
else:
breaker.record_failure()
raise Exception("API request failed")
except Exception as e:
breaker.record_failure()
raise e
四、代理配置：突破封禁的"隐身术"
- 代理池实战方案
import random
from requests.adapters import HTTPAdapter
class ProxyPool:
def init(self):
self.proxies = [
{"http": "http://1.1.1.1:8080", "https": "http://1.1.1.1:8080"},
{"http": "http://2.2.2.2:8080", "https": "http://2.2.2.2:8080"},
# 更多代理...
]
self.failed_proxies = set()
def get_proxy(self):
available_proxies = [p for p in self.proxies if p not in self.failed_proxies]
if not available_proxies:
raise Exception("No available proxies")
return random.choice(available_proxies)
def mark_failed(self, proxy):
self.failed_proxies.add(proxy)
使用示例
proxy_pool = ProxyPool()
session = requests.Session()
for _ in range(3): # 尝试3个不同代理
try:
proxy = proxy_pool.get_proxy()
response = session.get(
'https://api.sf-express.com/track',
proxies=proxy,
timeout=(3, 10)
)
if response.status_code == 200:
print("Success with proxy:", proxy)
break
else:
proxy_pool.mark_failed(proxy)
except Exception:
proxy_pool.mark_failed(proxy)
else:
print("All proxies failed")
- Tor代理配置（高匿名场景）
import requests
def make_tor_request(url):
proxies = {
'http': 'socks5h://127.0.0.1:9050',
'https': 'socks5h://127.0.0.1:9050'
}
try:
response = requests.get(url, proxies=proxies, timeout=(5, 15))
print("Tor出口IP:", response.json()['origin'])
return response
except Exception as e:
print("Tor请求失败:", e)
需提前安装Tor服务并启动
sudo apt install tor # Ubuntu系统
sudo service tor start
五、完整实战案例：顺丰物流追踪系统
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
import logging
from datetime import datetime
日志配置
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s'
)
class SFExpressTracker:
def init(self, app_key, app_secret):
self.app_key = app_key
self.app_secret = app_secret
self.session = self._create_session()
def _create_session(self):
"""创建带重试和超时的会话"""
retry_strategy = Retry(
total=3,
backoff_factor=1,
status_forcelist=[429, 500, 502, 503, 504],
allowed_methods=["GET", "POST"]
)
adapter = HTTPAdapter(max_retries=retry_strategy)
session = requests.Session()
session.mount("http://", adapter)
session.mount("https://", adapter)
return session
def _generate_sign(self, params):
"""生成API签名（简化版）"""
import hashlib
sorted_params = sorted(params.items(), key=lambda x: x[0])
raw_str = self.app_secret + ''.join([f"{k}{v}" for k, v in sorted_params]) + self.app_secret
return hashlib.md5(raw_str.encode()).hexdigest().upper()
def query_track(self, tracking_number):
"""查询物流轨迹"""
url = "https://bsp-ois.sf-express.com/bsp-ois/express/service/queryTrack"
params = {
"appKey": self.app_key,
"trackNumber": tracking_number,
"timestamp": str(int(datetime.now().timestamp()))
}
params["sign"] = self._generate_sign(params)
try:
response = self.session.get(
url,
params=params,
timeout=(3, 10) # 连接3秒，读取10秒
)
response.raise_for_status()
data = response.json()
if data.get('success'):
return data['data']['tracks']
else:
logging.warning(f"API返回错误: {data.get('errorMsg', '未知错误')}")
return None
except requests.exceptions.RequestException as e:
logging.error(f"请求失败: {str(e)}")
return None
使用示例
if name == "main":
tracker = SFExpressTracker(app_key="YOUR_APP_KEY", app_secret="YOUR_APP_SECRET")
tracks = tracker.query_track("SF123456789")
if tracks:
for step in tracks:
print(f"{step['acceptTime']} {step['acceptAddress']} - {step['remark']}")
六、常见问题Q&A
Q1：被网站封IP怎么办？
A：立即启用备用代理池，建议使用住宅代理（如站大爷IP代理），配合每请求更换IP策略。对于大规模爬取，可采用Tor网络或IP轮换中间件。
Q2：如何选择重试次数？
A：遵循"3次黄金法则"：
首次请求
指数退避重试2次（总计3次）
超过3次仍失败应触发熔断或人工干预
Q3：代理请求速度慢怎么解决？
A：
测试代理延迟：curl --socks5 127.0.0.1:9050 https://httpbin.org/ip
使用代理评分机制，淘汰高延迟代理
对代理请求添加User-Agent和常规请求头
Q4：如何记录重试日志？
A：扩展Retry类实现自定义日志：
from urllib3.util.retry import Retry
import logging
class LoggingRetry(Retry):
def init(self, args, **kwargs):
super().init(args, **kwargs)
self.logger = logging.getLogger(name)
def new(self, **kw):
self.logger.debug(f"Creating new retry adapter with params: {kw}")
return super().new(**kw)
def increment(self, method, url, response=None, error=None, **kwargs):
self.logger.warning(f"Retry attempt {self.total - self._remaining + 1} for {method} {url}")
return super().increment(method, url, response, error, **kwargs)
Q5：POST请求重试需要注意什么？
A：
确保请求是幂等的（如使用唯一请求ID）
在重试前检查响应是否已部分处理
考虑使用idempotency-key请求头（如Stripe API要求）
通过合理组合超时配置、智能重试和代理策略，可构建出应对各种异常场景的健壮API请求系统。实际开发中建议结合Prometheus监控重试率、失败率等指标，持续优化请求策略。
