影刀RPA 流程卡死的自动检测与恢复：心跳监控机制

影刀RPA 流程卡死的自动检测与恢复心跳监控机制RPA最怕的不是报错——报错了至少有日志你知道问题在哪。最怕的是卡死流程没报错也不往下执行就那么挂着。你回来一看已经卡了三个小时今天的数据全没处理。这篇文章讲怎么给流程加心跳——让它自己感知到自己卡住了然后自救。流程卡死的常见原因元素等待超时设得太长。设了300秒超时一个操作卡住就是5分钟。网页弹窗未处理。弹了个alert/confirm浏览器卡住了但流程以为还在跑。网络请求挂起。API请求发出去没有响应连接池耗尽。死循环。WHILE条件永远为真循环跑不出来。内存耗尽。大数据量操作导致内存溢出进程没崩但响应极慢。心跳机制的原理拼多多店群自动化上架方案心跳就是流程定期报告自己还活着。如果一个心跳周期内没有报告外部监控器就判定流程卡死了。流程内部 每完成一个关键步骤 → 写入心跳时间戳 流程外部监控器 每隔N秒检查心跳时间戳 → 如果超过X秒没更新 → 流程可能卡死了实现文件心跳importtimeimportosfromdatetimeimportdatetimeclassHeartbeat:文件心跳监控def__init__(self,heartbeat_file,interval10,max_gap60):self.heartbeat_fileheartbeat_file self.intervalinterval self.max_gapmax_gap# 超过这么长时间没有心跳就认为卡死defbeat(self,step_name):发送心跳msgf{datetime.now().isoformat()}|{step_name}withopen(self.heartbeat_file,w,encodingutf-8)asf:f.write(msg)defcheck_alive(self):检查是否还活着ifnotos.path.exists(self.heartbeat_file):returnFalse,心跳文件不存在withopen(self.heartbeat_file,r,encodingutf-8)asf:contentf.read().strip()last_time_strcontent.split(|)[0].strip()last_timedatetime.fromisoformat(last_time_str)gap(datetime.now()-last_time).total_seconds()ifgapself.max_gap:returnFalse,f最后心跳{gap:.0f}秒前超过{self.max_gap}秒阈值returnTrue,f存活{gap:.0f}秒前在流程的关键位置插入心跳主流程 heartbeat Heartbeat(rC:\RPA_data\heartbeat.txt, max_gap120) heartbeat.beat(流程启动) 【打开网页】 heartbeat.beat(页面加载完成) FOR EACH page IN pages: heartbeat.beat(f开始采集第{page}页)  采集数据... heartbeat.beat(f第{page}页采集完成) END FOR 【写入数据库】 heartbeat.beat(数据写入完成) heartbeat.beat(流程结束)外部监控器心跳文件只是个信号——还需要一个监控器来读这个信号并在异常时触发告警。可以用一个独立的流程或Python脚本来监控# monitor.py — 心跳监控器可以做成定时任务每分钟跑一次importosimporttimefromdatetimeimportdatetime heartbeat_filerC:\RPA_data\heartbeat.txtmax_gap120# 2分钟没心跳就告警ifnotos.path.exists(heartbeat_file):print([告警] 心跳文件不存在流程可能未启动)else:withopen(heartbeat_file,r)asf:contentf.read().strip()ifnotcontent:print([告警] 心跳文件为空)else:last_time_strcontent.split(|)[0].strip()last_timedatetime.fromisoformat(last_time_str)gap(datetime.now()-last_time).total_seconds()ifgapmax_gap:print(f[告警] 流程可能卡死最后心跳在{gap:.0f}秒前)# 发邮件/短信/企业微信通知else:print(f[正常] 最后心跳在{gap:.0f}秒前状态{content.split(|)[1].strip()})操作级别的超时保护光有心跳还不够每个可能卡住的操作都应该有超时TEMU店群如何管理运营importsignalimportfunctoolsdeftimeout(seconds):函数级别的超时装饰器defdecorator(func):functools.wraps(func)defwrapper(*args,**kwargs):# 注意signal只在主线程有效影刀的Python节点不一定支持# 这里提供另一种更通用的方式importthreading result[None]exception[None]deftarget():try:result[0]func(*args,**kwargs)exceptExceptionase:exception[0]e threadthreading.Thread(targettarget)thread.daemonTruethread.start()thread.join(timeoutseconds)ifthread.is_alive():raiseTimeoutError(f{func.__name__}执行超时{seconds}秒)ifexception[0]:raiseexception[0]returnresult[0]returnwrapperreturndecoratortimeout(30)defrisky_operation():# 可能卡住的操作pass监控看板可选如果你的RPA规模比较大十几个流程在跑可以考虑做一个简单的监控看板监控目录结构 heartbeat/ 订单采集.heartbeat 数据清洗.heartbeat 报表生成.heartbeat 邮件发送.heartbeat 监控汇总脚本 - 遍历heartbeat/目录 -  - 读每个文件的时间戳 - 输出监控面板 ✅ 订单采集12秒前 ✅ 数据清洗5秒前 ⚠️ 报表生成95秒前— 接近阈值 ❌ 邮件发送300秒前— 疑似卡死总结心跳机制就是流程每隔一段时间报告自己还活着。文件心跳是最简单的实现方式。心跳不是一个可选功能——任何需要长时间运行的RPA流程都应该有心跳。否则你永远不知道它是正常运行还是已经卡了两小时。作者林焱