影刀RPA API调用的重试退避与超时策略
作者:林焱
什么情况用这个
调用外部API时,有三种情况一定会遇到:超时、限流、服务暂时不可用。如果每种情况都让流程直接报错中断,那你每天有三分之一的时间在手动重启流程。
API调用的稳定性不靠"运气好不出错",靠的是重试策略和超时控制的合理设计。这篇文章给你一套工业级的API调用封装,让你的API请求能自动处理大部分常见问题。
怎么做
店群矩阵自动化突破运营极限!
基础封装
importrequestsimporttimeimportrandomfromrequests.adaptersimportHTTPAdapterfromurllib3.util.retryimportRetrydefcreate_retry_session(retries=3,backoff_factor=1):"""创建带重试的requests Session"""session=requests.Session()retry_strategy=Retry(total=retries,backoff_factor=backoff_factor,# 重试间隔: {backoff} * (2 ** (retry-1))status_forcelist=[429,500,502,503,504],# 这些状态码自动重试allowed_methods=["GET","POST"]# 允许重试的方法)adapter=HTTPAdapter(max_retries=retry_strategy)session.mount("http://",adapter)session.mount("https://",adapter)returnsession# 使用session=create_retry_session()resp=session.get("https://api.example.com/data",timeout=10)自定义重试逻辑
defapi_call_with_retry(url,method="GET",max_retries=3,**kwargs):"""带智能重试的API调用"""forattemptinrange(max_retries):try:resp=requests.request(method,url,timeout=kwargs.pop("timeout",10),**kwargs)# 限流 → 等服务器说的时长ifresp.status_code==429:wait=int(resp.headers.get("Retry-After",5))print(f"被限流,等待{wait}秒...")time.sleep(wait)continue# 服务器错误 → 退避重试ifresp.status_code>=500:delay=2**attempt+random.uniform(0,1)print(f"服务器错误{resp.status_code},{delay:.1f}秒后重试")time.sleep(delay)continuereturnrespexceptrequests.exceptions.Timeout:ifattempt==max_retries-1:raiseprint(f"超时,重试{attempt+1}/{max_retries}")time.sleep(2**attempt)exceptrequests.exceptions.ConnectionError:ifattempt==max_retries-1:raiseprint(f"连接失败,重试{attempt+1}/{max_retries}")time.sleep(3)超时控制
# 三层超时:连接超时 + 读取超时resp=requests.get(url,timeout=(5,30))# 连接5秒,读取30秒# 用Signal做硬超时(防止requests卡死)importsignaldefapi_call_with_hard_timeout(url,hard_timeout=60):"""带硬超时的API调用——到时间强制中断"""signal.alarm(hard_timeout)try:returnrequests.get(url,timeout=10)finally:signal.alarm(0)有什么坑
坑一:重试导致重复提交
现象:POST请求超时了但服务器其实已经处理了,重试导致重复创建。
解决:POST/DELETE等非幂等操作,默认不重试。或者用幂等键(idempotency key)防重。
坑二:错误的超时时间设置
temu店群自动化报活动案例
现象:timeout设了60秒,但平均响应时间就是1秒。超时请求要等60秒才知道失败了。
解决:超时时间 = 正常响应时间的5-10倍,而不是越大越好。
坑三:重试次数太多导致雪崩
现象:多个流程同时重试同一个挂掉的API,加重服务器负担。
解决:加入随机抖动(jitter),让重试时间分散:
delay=base_delay*(2**attempt)+random.uniform(0,3)总结:API调用的稳定性公式——合理超时 + 指数退避重试 + 限流遵从 + 非幂等不重试。把这四点封装成一个公用函数,所有API调用都走它。