影刀RPA 网络超时与重试机制请求稳定性保障作者林焱什么情况用你的影刀流程采集数据时偶尔网络波动导致请求失败整个流程就中断了有些API接口偶尔返回500错误但刷新一下就好了你想让流程在网络不稳定时自动重试而不是直接报错退出网络超时与重试机制是保障流程稳定性的核心。网络请求不可能100%成功必须有超时控制和失败重试策略。没有重试机制的流程在生产环境中根本跑不稳。本文讲清楚怎么设置合理的超时时间、怎么实现多种重试策略、怎么避免重试导致的问题。怎么做一、超时设置importrequests# 基本超时设置 # 单一超时值连接和读取都用这个时间responserequests.get(url,timeout10)# 分别设置连接超时和读取超时# (连接超时, 读取超时)responserequests.get(url,timeout(5,30))# 连接5秒读取30秒[video(video-cUSsdyfD-1784656707680)(type-csdn)(url-https://live.csdn.net/v/embed/526818)(image-https://v-blog.csdnimg.cn/asset/582d14c3bd0451c5399cd990b56e2a0d/cover/Cover0.jpg)(title-拼多多店群自动化报活动上架)]# 不同场景的超时建议 TIMEOUT_PRESETS{fast:(3,10),# 内网API、快速查询normal:(5,30),# 普通网页请求slow:(10,60),# 大数据量、慢接口download:(10,300),# 文件下载upload:(10,120),# 文件上传}二、基本重试机制importrequestsimporttimedefrequest_with_retry(url,max_retries3,timeout30,**kwargs):带重试的请求forattemptinrange(1,max_retries1):try:responserequests.get(url,timeouttimeout,**kwargs)# 检查状态码ifresponse.status_code200:returnresponse# 服务器错误5xx可以重试ifresponse.status_code500:print(f第{attempt}次请求服务器错误{response.status_code})ifattemptmax_retries:waitattempt*2# 递增等待print(f等待{wait}秒后重试...)time.sleep(wait)continue# 客户端错误4xx不重试response.raise_for_status()exceptrequests.Timeout:print(f第{attempt}次请求超时)ifattemptmax_retries:time.sleep(attempt*2)exceptrequests.ConnectionError:print(f第{attempt}次请求连接失败)ifattemptmax_retries:time.sleep(attempt*2)exceptExceptionase:print(f第{attempt}次请求异常{e})ifattemptmax_retries:time.sleep(attempt*2)print(f全部{max_retries}次重试失败)returnNone三、多种重试策略importtimeimportrandom# 策略1固定间隔重试 defretry_fixed_interval(func,max_retries3,interval2):固定间隔重试forattemptinrange(1,max_retries1):try:returnfunc()exceptExceptionase:print(f第{attempt}次失败:{e})ifattemptmax_retries:time.sleep(interval)raiseException(f重试{max_retries}次后仍失败)# 策略2递增间隔重试 defretry_incremental(func,max_retries5,base_delay1):递增间隔重试1s, 2s, 3s, 4s, 5sforattemptinrange(1,max_retries1):try:returnfunc()exceptExceptionase:print(f第{attempt}次失败:{e})ifattemptmax_retries:delaybase_delay*attemptprint(f等待{delay}秒后重试...)time.sleep(delay)raiseException(f重试{max_retries}次后仍失败)# 策略3指数退避重试推荐 defretry_exponential_backoff(func,max_retries5,base_delay1,max_delay60): 指数退避重试1s, 2s, 4s, 8s, 16s 最适合网络请求场景 forattemptinrange(1,max_retries1):try:returnfunc()exceptExceptionase:print(f第{attempt}次失败:{e})ifattemptmax_retries:delaymin(base_delay*(2**(attempt-1)),max_delay)# 加随机抖动避免同时重试jitterrandom.uniform(0,delay*0.1)total_delaydelayjitterprint(f等待{total_delay:.1f}秒后重试...)time.sleep(total_delay)raiseException(f重试{max_retries}次后仍失败)# 策略4条件重试 defretry_with_conditions(func,max_retries3,retry_on_exceptionsNone): 只对特定异常重试 retry_on_exceptionsretry_on_exceptionsor[Exception]forattemptinrange(1,max_retries1):try:returnfunc()exceptExceptionase:should_retryany(isinstance(e,exc)forexcinretry_on_exceptions)ifshould_retryandattemptmax_retries:print(f第{attempt}次失败可重试:{type(e).__name__}:{e})time.sleep(attempt*2)else:print(f第{attempt}次失败不可重试:{type(e).__name__}:{e})raiseraiseException(f重试{max_retries}次后仍失败)# 使用示例importrequests resultretry_exponential_backoff(lambda:requests.get(https://api.example.com/data,timeout10).json(),max_retries5,base_delay1)# 只对网络异常重试不对JSON解析错误重试resultretry_with_conditions(lambda:requests.get(https://api.example.com/data,timeout10).json(),max_retries3,retry_on_exceptions[requests.Timeout,requests.ConnectionError])四、使用urllib3的Retry机制fromurllib3.util.retryimportRetryfromrequests.adaptersimportHTTPAdapterimportrequestsdefcreate_session_with_retry(total_retries3,backoff_factor1,status_forcelistNone,allowed_methodsNone): 创建带自动重试的requests Session 这是requests库内置的重试机制比手动重试更底层更可靠 sessionrequests.Session()retryRetry(totaltotal_retries,# 总重试次数backoff_factorbackoff_factor,# 退避因子第n次重试等待 backoff_factor * (2^(n-1)) 秒status_forceliststatus_forcelistor[500,502,503,504],# 对这些状态码重试allowed_methodsallowed_methodsor[GET,PUT,DELETE],# 对这些HTTP方法重试raise_on_statusFalse,# 重试次数用完后不抛异常返回响应)adapterHTTPAdapter(max_retriesretry)session.mount(http://,adapter)session.mount(https://,adapter)returnsession# 使用sessioncreate_session_with_retry(total_retries3,backoff_factor1,# 等待 1s, 2s, 4sstatus_forcelist[500,502,503,504])# 这个session会自动重试responsesession.get(https://api.example.com/data,timeout30)五、完整的稳定请求封装importrequestsimporttimeimportrandomfromurllib3.util.retryimportRetryfromrequests.adaptersimportHTTPAdapterclassStableHTTPClient:稳定的HTTP客户端内置超时重试错误处理def__init__(self,timeout(5,30),max_retries3,backoff_factor1):self.timeouttimeout self.sessionrequests.Session()# 配置自动重试retryRetry(totalmax_retries,backoff_factorbackoff_factor,status_forcelist[500,502,503,504],allowed_methods[GET,POST],)adapterHTTPAdapter(max_retriesretry)self.session.mount(http://,adapter)self.session.mount(https://,adapter)defrequest(self,method,url,**kwargs):发送请求kwargs.setdefault(timeout,self.timeout)try:responseself.session.request(method,url,**kwargs)ifresponse.status_code200:return{success:True,data:response,status:200}elifresponse.status_code429:# 被限流等待后重试retry_afterint(response.headers.get(Retry-After,10))print(f被限流等待{retry_after}秒...)time.sleep(retry_after)returnself.request(method,url,**kwargs)# 递归重试else:return{success:False,status:response.status_code,error:fHTTP{response.status_code}}exceptrequests.Timeout:return{success:False,error:请求超时}exceptrequests.ConnectionError:return{success:False,error:连接失败}exceptExceptionase:return{success:False,error:str(e)}defget(self,url,paramsNone):returnself.request(GET,url,paramsparams)defpost(self,url,dataNone,jsonNone):returnself.request(POST,url,datadata,jsonjson)# 使用clientStableHTTPClient(timeout(5,30),max_retries3)resultclient.get(https://api.example.com/data,params{page:1})ifresult[success]:dataresult[data].json()print(f获取数据成功:{data})else:print(f获取数据失败:{result[error]})六、采集场景的重试策略defscrape_with_retry(url,scrape_func,max_retries3): 采集重试的组合 scrape_func: 采集函数接收response返回数据 clientStableHTTPClient(max_retriesmax_retries)forattemptinrange(1,max_retries1):resultclient.get(url)ifresult[success]:try:datascrape_func(result[data])return{success:True,data:data}exceptExceptionase:print(f第{attempt}次解析失败:{e})ifattemptmax_retries:time.sleep(attempt*2)else:print(f第{attempt}次请求失败:{result[error]})ifattemptmax_retries:time.sleep(attempt*2)return{success:False,error:全部重试失败}# 使用defparse_products(response):解析商品数据dataresponse.json()productsdata.get(items,[])ifnotproducts:raiseValueError(商品列表为空)returnproducts resultscrape_with_retry(https://api.example.com/products?page1,parse_products,max_retries3)有什么坑坑1重试导致数据重复写入现象POST请求超时了重试后成功但数据库里多了两条相同的数据。原因第一次请求可能实际已经到达服务器并处理成功了只是响应没返回网络超时。重试时又提交了一次导致重复。解决对POST/PUT等写操作设计幂等性——用唯一请求IDrequest_id服务器根据request_id判断是否已处理。或者在重试前先查询数据是否已存在。GET请求天然幂等可以安全重试。坑2重试次数太多拖慢流程现象每个请求重试5次每次等8秒一个失败请求要等40秒才放弃流程严重变慢。原因重试次数和等待时间设置不合理对注定失败的请求还在反复重试。解决根据场景调整重试参数。临时性错误超时、503可以多重试永久性错误404、401不重试。设置总超时时间上限——即使重试也不能超过某个总时间。对于采集流程一个页面失败就跳过不要影响整体进度。TEMU店群矩阵自动化运营核价报活动坑3重试时没有切换代理或UA现象被网站封IP后重试每次重试都用同一个IP和UA全部失败。原因重试时没有改变任何请求特征网站继续拒绝。解决重试时切换代理IP和User-Agent。让每次重试看起来像不同的用户访问forattemptinrange(max_retries):headers{User-Agent:get_random_ua()}proxiesget_random_proxy()responserequests.get(url,headersheaders,proxiesproxies,timeout10)坑4递归重试导致栈溢出现象用递归方式实现重试深度太大会栈溢出。解决重试用循环for/while实现不要用递归。如果必须递归设最大深度限制。坑5影刀的【尝试-异常】和Python重试混用现象在影刀中用【尝试-异常】包裹了【执行Python代码】Python代码内部也有重试逻辑导致重试次数翻倍。解决重试逻辑只在一个层面实现——要么在Python代码块内用retry逻辑要么在影刀流程层面用【尝试-异常】【循环】实现。不要两层都做重试。推荐在Python代码块内处理重试影刀层面只做异常捕获和流程控制。