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