影刀RPA 网络超时与重试策略：打造高可用的自动化流程

影刀RPA 网络超时与重试策略打造高可用的自动化流程作者林焱网络请求是RPA流程中最不稳定的环节——超时、断连、DNS解析失败、服务器500错误……这些问题不是会不会发生而是什么时候发生。唯一正确的态度是默认网络不可靠做好重试和降级。基础重试模板importrequestsimporttimedefrequest_with_retry(url,max_retries3,**kwargs):带重试的HTTP请求forattemptinrange(max_retries):try:resprequests.get(url,timeout10,**kwargs)# 服务器错误重试ifresp.status_code500:print(f服务器错误({resp.status_code})第{attempt1}次重试...)time.sleep(2**attempt)# 指数退避continue# 限速重试但要等更久ifresp.status_code429:retry_afterint(resp.headers.get(Retry-After,5))print(f被限速等待{retry_after}秒...)time.sleep(retry_after)continuereturnresp# 成功或客户端错误exceptrequests.exceptions.Timeout:print(f超时第{attempt1}次重试...)time.sleep(2**attempt)exceptrequests.exceptions.ConnectionError:print(f连接失败第{attempt1}次重试...)time.sleep(3)# 连接问题等一下再试exceptExceptionase:print(f未知错误:{e})ifattemptmax_retries-1:raise# 最后一次抛出异常raiseException(f请求失败已重试{max_retries}次:{url})指数退避 vs 固定延时拼多多店群自动化上架方案# 不好固定延时foriinrange(3):try:returnrequests.get(url)except:time.sleep(3)# 每次都等3秒# 好指数退避越来越久foriinrange(3):try:returnrequests.get(url)except:waitmin(2**i,30)# 1秒 → 2秒 → 4秒 → 8秒...最多30秒time.sleep(wait)超时控制# 错误不设超时resprequests.get(url)# 可能永远卡住# 正确设连接超时和读取超时resprequests.get(url,timeout(5,30))# 连接5秒读取30秒# 不同场景的超时策略TIMEOUTS{quick:(3,10),# 快速查询normal:(5,30),# 常规请求download:(10,120),# 文件下载}断路器模式如果连续失败N次短时间内不再尝试避免雪崩importtimeclassCircuitBreaker:def__init__(self,failure_threshold5,recovery_timeout60):self.failure_count0self.thresholdfailure_threshold self.recovery_timeoutrecovery_timeout self.last_failure_time0self.stateCLOSED# CLOSED / OPEN / HALF_OPENdefcall(self,func,*args,**kwargs):ifself.stateOPEN:iftime.time()-self.last_failure_timeself.recovery_timeout:self.stateHALF_OPENprint(断路器进入半开状态尝试恢复...)else:raiseException(断路器已打开拒绝请求)try:resultfunc(*args,**kwargs)self.failure_count0self.stateCLOSEDreturnresultexceptExceptionase:self.failure_count1self.last_failure_timetime.time()ifself.failure_countself.threshold:self.stateOPENprint(f连续失败{self.failure_count}次断路器打开)raisee# 使用breakerCircuitBreaker(failure_threshold3,recovery_timeout30)forurlinurls:try:respbreaker.call(requests.get,url,timeout10)process(resp)except:print(f跳过:{url}断路器/超时)网络自检defnetwork_health_check():启动前检查网络状态checks{百度:https://www.baidu.com,目标站点:https://target-site.com}results{}forname,urlinchecks.items():try:resprequests.get(url,timeout5)results[name]f✓ ({resp.elapsed.total_seconds():.2f}s)except:results[name]✗ 不通all_okall(✓invforvinresults.values())returnall_ok,results踩坑实录坑1重试导致重复操作GET请求重试是幂等的重复请求不影响数据但POST请求重试可能导致重复创建记录。对非幂等的请求重试前先检查是否已经创建成功TEMU店群如何管理运营ifresp.status_code500:# 检查是否已经创建了某些服务器收到了请求但返回了500check_resprequests.get(f{url}/{id})ifcheck_resp.status_code200:returncheck_resp# 其实已经创建成功了坑2重试太多次导致流程卡死每个URL重试3次、每次等8秒100个URL如果全失败就是40分钟。设置总超时importtime starttime.time()TOTAL_TIMEOUT300# 整个流程最多5分钟forurlinurls:iftime.time()-startTOTAL_TIMEOUT:print(流程总超时停止)breakrequest_with_retry(url)坑3所有重试都用一样的策略不同错误的处理方式不同DNS解析失败 → 等久一点DNS缓存还没更新连接超时 → 马上重试可能是临时网络波动429限速 → 等Retry-After指定的时间500服务器错误 → 指数退避服务器可能正在恢复写在最后网络不稳定是常态不是异常。建好重试和容错机制让你的影刀流程能在网络抖动时自动恢复而不是一碰就挂。花了半小时写重试逻辑未来省你无数次半夜爬起来手动重启流程。