影刀RPA 定时任务的容错设计：失败重试与降级通知

作者：林焱

什么情况用这个

你设了每天早8点自动跑的数据采集流程。某天早上你打开电脑，发现流程凌晨2点就挂了，原因是数据库临时维护，连接失败了。到现在已经过去6个小时，没人知道流程失败了，数据没更新，业务方在等你。

定时任务的容错不是"不出错"，而是"出错了一定有人知道，并且自动尝试恢复"。这篇文章教你给定时任务加上自愈和告警能力。

怎么做

失败重试

importtimefromdatetimeimportdatetimedefrun_with_retry(func,max_retries=3,retry_delay=300):"""定时任务重试包装器"""forattemptinrange(max_retries):try:func()returnTrueexceptExceptionase:ifattempt==max_retries-1:# 最后一次也失败了 → 告警send_alert(f"任务{func.__name__}重试{max_retries}次后仍然失败:{e}")raiseprint(f"第{attempt+1}次失败，{retry_delay}秒后重试:{e}")time.sleep(retry_delay)

健康检查

defhealth_check():"""运行前检查依赖是否可用"""checks={"网络":check_network(),"数据库":check_database(),"磁盘空间":check_disk_space(min_gb=1),"源文件存在":os.path.exists("D:/data/source.xlsx"),}![在这里插入图片描述](https://i-blog.csdnimg.cn/direct/3105098cf0cb4a39a0443d299022b094.png#pic_center)failed=[kfork,vinchecks.items()ifnotv]iffailed:send_alert(f"健康检查失败:{', '.join(failed)}")returnFalsereturnTrue

结果验证

defvalidate_output():"""运行后验证结果"""# 检查输出文件是否生成且非空ifnotos.path.exists("D:/output/report.xlsx"):returnFalse,"输出文件未生成"ifos.path.getsize("D:/output/report.xlsx")<1024:returnFalse,"输出文件过小（可能是空数据）"returnTrue,""

有什么坑

坑一：只告警不自愈

现象：每次失败都发通知，人工手动重跑，但从没想过让流程自动重试。

解决：先自动重试2-3次，重试全失败了再告警。

坑二：告警疲劳

现象：每天收20条告警消息，后来看都不看了。

解决：告警分级——ERROR级别即时推送，WARNING级别汇总日报。同类型错误1小时内只告警一次。

坑三：没有最终兜底

现象：流程失败、自动重试也失败、告警发了但没人看。

解决：加一个"最终防线"——邮件抄送领导、钉钉电话通知、或其他确保一定有人看到的方式。

总结：定时任务容错 =运行前健康检查 + 失败自动重试 + 结果验证 + 分级告警。不求完美运行100%的成功率，只求出问题了10分钟内有人知道、30分钟内有人处理。