去年在做一块 ESP32-S3 的室外数据采集节点时,遇到一个非常头疼的问题:设备偶尔会莫名其妙地卡死,不是死机那种彻底没反应,而是主循环还在跑,但网络任务和传感器读取任务已经完全不响应了。重启之后能好一阵子,过几天又犯。当时排查了很久,最后才发现是某个底层驱动在极端情况下进入了不可恢复的阻塞状态。从那以后我意识到,嵌入式开发里真正让人头疼的从来不是“跑不起来”,而是“跑着跑着不干活了”。也就是从那时起,我开始认真研究软件看门狗的各种玩法,尤其是带恢复机制的那种。
这篇文章就想把我这段时间在 MicroPython 下实现软件看门狗的经验完整地梳理一遍。我会从最基础的概念讲起,包括为什么在 MicroPython 这种带垃圾回收和异常机制的环境里依然需要看门狗、软件看门狗和硬件看门狗到底怎么选、如何设计一个真正能“恢复现场”而不是只会重启的看门狗机制,以及我在实际项目中踩过的坑和最终采用的完整代码方案。无论你是在做 ESP32、RP2040 还是其他支持 MicroPython 的开发板,这套思路基本都能直接用。
1. 为什么 MicroPython 项目里也需要看门狗
很多刚开始做 MicroPython 开发的朋友会有一个错觉:Python 不是有异常处理吗?不是有 try-except 吗?就算程序崩了,最多就是报个错,重新运行一下不就行了。更何况 MicroPython 还有垃圾回收机制,看起来比裸机 C 开发要安全得多。但实际上,这种想法在真实项目中是站不住脚的。
1.1 Python 世界的“假安全”
MicroPython 确实比传统的 C 固件开发多了一层保护,比如数组越界会抛出 IndexError,除零会抛出 ZeroDivisionError,指针问题也基本被隐藏了。但问题在于,嵌入式环境里的很多故障根本不会通过异常暴露出来。最典型的就是死循环:某个 while 条件因为外部状态变化永远无法满足,代码就卡在那里了,CPU 占用率拉满,但所有其他任务都在饿死。这时候不会有任何异常抛出,你甚至不知道系统已经挂了。
另一个常见情况是阻塞调用超时。比如你用 I2C 读取一个传感器,传感器因为硬件故障把 SDA 线拉低了,如果你的驱动库实现里没有超时机制,那 read 操作会一直等下去。在 PC 上你可能还有操作系统帮你调度,但在单片机上,这就是彻底的死等。
我做个比喻:异常处理相当于你出门前检查了煤气灶,确保没有明火隐患。但看门狗是隔壁邻居,他不管你出门时煤气灶有没有关,只负责每过一阵子确认一下你家还有没有动静——如果连续很长时间没动静,他就直接踹门进来帮你处理了。
1.2 从 Python 移植到 MicroPython 时最容易丢掉的保护
在 PC 端写 Python 程序,如果一个函数卡死了,操作系统不会让整个系统崩溃,因为每个进程是隔离的。但在 MicroPython 的开发板上,整个应用程序就跑在单个进程中,没有操作系统帮你兜底。一旦主循环卡死,MicroPython 的调度器也停了,定时器回调、网络回调、异步任务全部瘫痪。
更隐蔽的是,MicroPython 的底层固件和你的 Python 应用层是两层东西。固件层可能有 bug,驱动层可能有 bug,这些都不是 Python 层的 try-except 能捕获到的。比如我之前遇到过 ESP32 的 WiFi 驱动在特定重启时序下会进入一个异常状态,表现为 connect 方法永远不返回,而且这个异常发生在 C 固件层,Python 代码根本感知不到。
所以哪怕你用 MicroPython,哪怕你做了完善的异常处理,依然需要一个“最后防线”来兜底。这个防线就是看门狗。
2. 软件看门狗与硬件看门狗的选择逻辑
很多初学者会有个误区,觉得看门狗就只是调一个 API 的事。其实在真正设计系统时,首先要回答的问题是:我用硬件看门狗还是软件看门狗,还是两者都用?
2.1 硬件看门狗的本质与局限
硬件看门狗是芯片内部一个独立计数器,需要在规定时间内“喂狗”,否则计数器溢出就会强制复位整个芯片。它的优点是极其可靠——即使 CPU 跑飞了、中断系统崩溃了,只要喂狗操作没执行,芯片就会被拉回来重启。
但硬件看门狗有个很大的问题:它太“粗”了。它只关心“喂了没有”,不关心“喂得对不对”。如果你的主循环因为某个 bug 进入了死循环,但恰恰这个死循环里还在执行喂狗操作,那硬件看门狗形同虚设。更常见的情况是,你的主逻辑已经卡死了,但一个高优先级的中断服务函数还在运行,而这个中断函数里碰巧有喂狗代码——系统实际已经半死不活,但看门狗依然被喂得好好的,永远不会触发复位。
另一个问题是硬件看门狗几乎没有“恢复”的能力。它触发了就是整个系统复位,所有 RAM 数据全部丢失,所有外设状态全部重新初始化。对很多需要保持业务连续性的场景来说,这种“一刀切”是不可接受的。
2.2 软件看门狗的“柔性”价值
软件看门狗的思路就不一样了。它可以看作是运行在应用层的一个“监控线程”,定期检查各个任务是否在预期时间内完成心跳更新。如果某个任务超时了,软件看门狗可以做出比“重启”更精细的操作:比如只重启那个失联的任务模块、重新初始化某个外设、切换到一个降级运行模式、保留当前 RAM 中的关键数据然后再复位。
软件看门狗的本质价值就是四个字:分级恢复。它能根据故障的严重程度选择不同的恢复策略,而不是无论什么都直接重启。
但也要承认,软件看门狗有一个天然的短板:它本身也是软件,它自己也可能被卡死。如果主循环都卡死了,软件看门狗还怎么工作?所以最稳妥的方案是“软件看门狗 + 硬件看门狗”组合:软件看门狗负责细粒度监控和分级恢复,硬件看门狗作为最后的粗粒度保险。
2.3 何时可以只用软件看门狗
在实际项目中,软件看门狗单独使用也是可行的,但有一些前提条件。首先是你的主循环必须能保证周期性运行,因为软件看门狗通常依赖主循环或定时器来执行检查。如果你的系统有 RTOS 或者 MicroPython 的 asyncio 支持,那软件看门狗的可靠性会高很多,因为即使某个 task 卡死了,其他 task 还能运行。
其次是故障恢复目标不是“完全自动恢复”,而是“可感知、可上报”。比如某些数据采集设备,卡死后只要能发一条 MQTT 消息说“我挂了”,或者把故障状态写入非易失存储,然后重启,这就够了。这种情况下软件看门狗完全能胜任,而且比硬件看门狗更灵活。
如果你采用的是“裸奔”式主循环架构(就是没有 RTOS、没有 asyncio,一个 while True 跑到底),那我建议至少要加一个定时器中断作为软件看门狗的“心跳源”,否则主循环卡死了软件看门狗自己也跑不了。
3. MicroPython 看门狗 API 及基础实现
先来看 MicroPython 官方标准库里的基础看门狗接口:machine.WDT。这个类在大多数支持 MicroPython 的开发板上都有实现(ESP32、RP2040、STM32 等),用法非常简单。
from machine import WDT # 创建看门狗实例,超时时间 5 秒 wdt = WDT(timeout=5000) # 单位是毫秒 # 主循环中周期性喂狗 while True: # ... 业务逻辑 ... wdt.feed()注意,MicroPython 的machine.WDT实现的是硬件看门狗功能。也就是说你一旦创建了 WDT 实例,喂狗周期就必须小于 timeout,否则系统会被强制复位。这是硬件行为,不是 Python 层的软逻辑。
3.1 network 场景下 WDT 的误触发现象
用machine.WDT的过程中我踩过一个大坑:如果你的主循环里用了阻塞式的网络请求,比如requests.get(),而且没有设置超时,那么当网络异常时这个请求可能阻塞几十秒甚至更久,硬件看门狗就会因为没被喂到而触发复位。
你可能会说:网络请求不是有 timeout 参数吗?确实有,但有些库在底层 TCP 重传机制里,timeout 并不可靠。我遇到过设置了 timeout=10 但实际阻塞了 45 秒的情况。所以如果你在带网络功能的项目里用machine.WDT,一定要确保喂狗操作发生在主循环最外层,并且在所有可能长时间阻塞的调用前后都至少要保证主循环有返回的机会。
一个比较稳妥的做法是把喂狗操作放在定时器中断里:
from machine import Timer, WDT wdt = WDT(timeout=5000) def feed_wdt(timer): wdt.feed() # 每 2 秒喂一次 timer = Timer(0) timer.init(period=2000, mode=Timer.PERIODIC, callback=feed_wdt)这样即使主循环被某个阻塞调用卡住了,定时器中断依然可以喂狗,避免误复位。但这就回到了前面说的问题:既然喂狗操作在中断里,主逻辑卡死了 WDT 也不会触发——硬件看门狗失效了。所以这不是一个完美的方案,只是一个权衡。
3.2 在 MicroPython 中模拟硬件看门狗触发
有些开发板(比如某些 RP2040 的移植版本)对machine.WDT的支持并不完整,或者你需要在启动早期就使用看门狗但那时 WDT 对象还没初始化好。这时候可以用一种“假看门狗”的方式在 Python 层模拟:
import time class FakeWDT: def __init__(self, timeout_ms): self.timeout_ms = timeout_ms self.last_feed_time = time.ticks_ms() def feed(self): self.last_feed_time = time.ticks_ms() def check(self): if time.ticks_diff(time.ticks_ms(), self.last_feed_time) > self.timeout_ms: raise SystemExit("WDT timeout")然后在主循环里手动检查:
wdt = FakeWDT(5000) while True: wdt.check() # ... 业务逻辑 ... try: wdt.feed() except SystemExit: machine.reset()这种方式的优点是你完全控制检查时机,可以在检查之前先尝试做现场保存。缺点是它依赖主循环本身没有卡死。所以它更适合用来做“业务逻辑级别”的看门狗,而不是系统级别的救命稻草。
4. 带恢复机制的软件看门狗核心设计
前面讲了基础看门狗的各种形态,但说实话,这些距离这篇文章标题里的“带恢复机制”还有不小的距离。真正的软件看门狗,不应该只是“超时了就重启”,而应该具备诊断、恢复、升级、降级等一系列能力。这一节我就展开讲讲我是怎么设计的。
4.1 设计目标:从“重启”到“康复”
在设计带恢复机制的软件看门狗之前,先要想清楚一件事:重启不是目的,让系统恢复健康才是目的。
把这个目标拆解一下,软件看门狗要回答这几个问题:
- 系统现在处于什么状态?是轻微异常还是严重异常?
- 有没有可能在不重启的情况下恢复?如果能,恢复到什么程度?
- 如果必须重启,重启后如何避免立刻再次陷入同样的故障?
- 故障信息怎么记录,方便定位根因?
围绕这些问题,我设计了一个三层恢复架构:
- 第一层:任务级恢复。每个业务模块(传感器读取、网络通信、显示刷新等)都有独立的心跳。某个模块超时了,先尝试单独重置该模块,而不是动整个系统。
- 第二层:系统级软恢复。如果多个模块同时超时,或者单个模块连续多次重置仍然异常,就主动做一次“软复位”。软复位包括重新初始化关键外设、清除可能的内存垃圾、重置状态机,但保留全局配置和不必要的数据。
- 第三层:硬件级复位。前两层都失败,或者检测到系统已完全失控(连看门狗线程本身都无法运行),就触发硬件复位。
4.2 心跳监控表:软件看门狗的核心数据结构
要实现上面说的分级恢复,核心是要有一个“心跳监控表”,记录每个注册模块的状态。我使用字典来管理:
class SoftwareWatchdog: def __init__(self): self.heartbeats = {} # 模块名 -> 心跳记录 self.failure_counts = {} # 模块名 -> 连续失败次数 self.thresholds = {} # 模块名 -> (超时阈值, 最大连续失败次数) def register(self, name, timeout_ms=5000, max_failures=3): """注册一个需要监控的模块""" self.heartbeats[name] = { 'last_heartbeat': time.ticks_ms(), 'status': 'healthy' } self.thresholds[name] = (timeout_ms, max_failures) self.failure_counts[name] = 0 def heartbeat(self, name): """模块主动上报心跳""" if name in self.heartbeats: self.heartbeats[name]['last_heartbeat'] = time.ticks_ms() self.heartbeats[name]['status'] = 'healthy' self.failure_counts[name] = 0 def check(self): """周期性检查所有模块的心跳状态""" now = time.ticks_ms() alerts = [] for name, hb in self.heartbeats.items(): elapsed = time.ticks_diff(now, hb['last_heartbeat']) timeout_ms, max_failures = self.thresholds[name] if elapsed > timeout_ms: self.failure_counts[name] += 1 hb['status'] = 'timeout' if self.failure_counts[name] >= max_failures: alerts.append((name, 'critical')) else: self.failure_counts[name] = 0 return alerts这个结构的好处是:
- 每个模块独立设置超时阈值,传感器读取可能需要 2 秒,网络注册可能需要 10 秒,互不影响。
- 连续失败计数给了系统一个“容错空间”,避免单次偶发故障就触发大动作。
- 状态字段可以方便地扩展,比如加入
degraded状态表示模块还在运行但质量下降了。
4.3 恢复策略分发器:怎么决定“下一步做什么”
当check()发现问题后,需要一个恢复策略分发器来决定具体执行什么操作。我的实现思路是这样的:
class WatchdogRecovery: def __init__(self): self.recovery_steps = {} def register_recovery(self, module_name, recovery_callable): """为模块注册恢复函数""" self.recovery_steps[module_name] = recovery_callable def recover(self, module_name): """执行模块级恢复""" if module_name in self.recovery_steps: print(f"[WDG] Attempting recovery for {module_name}") try: self.recovery_steps[module_name]() return True except Exception as e: print(f"[WDG] Recovery failed: {e}") return False return False每个模块在注册时,同时注册一个“恢复函数”。这个函数负责重新初始化模块所需的外设、重置模块内部状态等。例如传感器模块的恢复函数可能是:
def sensor_recovery(): # 重新初始化 I2C 总线 i2c = I2C(0, scl=Pin(22), sda=Pin(21)) # 重新搜索传感器设备地址 devices = i2c.scan() if 0x76 in devices: print("[Sensor] I2C device found, reinit OK") else: raise RuntimeError("Sensor still missing")恢复失败后,才进入系统级软复位流程。这样的分级设计能最大程度减少无谓重启带来的业务中断。
5. 完整实战:MicroPython 看门狗子系统源码详解
这一节给出我实际项目里在用的完整看门狗子系统代码。这个代码我已经在 ESP32-S3 和 Raspberry Pi Pico W 上分别测试过,可以直接跑起来做实验。
5.1 主控文件 wdt_system.py
""" wdt_system.py - 带恢复机制的软件看门狗子系统 适用于 MicroPython 环境 (ESP32 / RP2040 etc.) 特性: 1. 多模块独立心跳监控 2. 连续失败计数与容错 3. 分级恢复策略(模块级恢复 -> 系统软复位 -> 硬件复位) 4. 故障日志记录 """ import time import machine import json import os class SoftwareWatchdog: def __init__(self, hw_wdt_timeout_ms=10000): self.heartbeats = {} self.failure_counts = {} self.thresholds = {} self.recovery_handlers = {} self.system_status = "healthy" self.fault_log = [] self.hw_wdt_timeout_ms = hw_wdt_timeout_ms self.hw_wdt = None def enable_hw_wdt(self): """启用硬件看门狗作为最终保险""" try: from machine import WDT self.hw_wdt = WDT(timeout=self.hw_wdt_timeout_ms) print("[WDG] Hardware WDT enabled, timeout = {} ms".format(self.hw_wdt_timeout_ms)) except Exception as e: print("[WDG] Failed to enable HW WDT: {}".format(e)) def register(self, name, timeout_ms=5000, max_failures=3): """注册监控模块 Args: name: 模块名称 timeout_ms: 心跳超时阈值 max_failures: 连续超时多少次后判定为 critical """ self.heartbeats[name] = { "last_heartbeat": time.ticks_ms(), "status": "healthy" } self.failure_counts[name] = 0 self.thresholds[name] = (timeout_ms, max_failures) print(f"[WDG] Registered module: {name}, timeout={timeout_ms}ms, max_failures={max_failures}") def register_recovery(self, module_name, handler): """为模块注册恢复函数""" self.recovery_handlers[module_name] = handler print(f"[WDG] Recovery handler registered for {module_name}") def heartbeat(self, name): """模块上报心跳""" if name in self.heartbeats: self.heartbeats[name]["last_heartbeat"] = time.ticks_ms() self.heartbeats[name]["status"] = "healthy" self.failure_counts[name] = 0 else: print(f"[WDG] Unknown module heartbeat ignored: {name}") def check_and_recover(self): """检查所有模块状态并执行恢复策略 返回当前系统状态(healthy / warning / critical) """ if self.hw_wdt is not None: self.hw_wdt.feed() now = time.ticks_ms() alerts = [] for name, hb in self.heartbeats.items(): elapsed = time.ticks_diff(now, hb["last_heartbeat"]) timeout_ms, max_failures = self.thresholds[name] if elapsed <= timeout_ms: self.failure_counts[name] = 0 hb["status"] = "healthy" continue # 模块心跳超时 self.failure_counts[name] += 1 hb["status"] = "timeout" print(f"[WDG] Module '{name}' heartbeat timeout ({elapsed}ms > {timeout_ms}ms), " f"failure count = {self.failure_counts[name]}") if self.failure_counts[name] >= max_failures: alerts.append((name, "critical")) if not alerts: self.system_status = "healthy" return "healthy" # 有模块达到 critical 状态,执行恢复策略 self.system_status = "critical" for name, level in alerts: self._execute_recovery(name) # 如果还没有恢复,触发系统级恢复 if self._all_critical_modules_recovered(): return "healthy" return "critical" def _execute_recovery(self, module_name): """先执行模块级恢复,失败则记录故障日志""" print(f"[WDG] Executing recovery for module: {module_name}") handler = self.recovery_handlers.get(module_name) if handler is None: print(f"[WDG] No recovery handler for {module_name}, will trigger system reset") self._log_fault(module_name, "no_recovery_handler") return False try: handler() print(f"[WDG] Module {module_name} recovered successfully") self.failure_counts[module_name] = 0 self.heartbeats[module_name]["last_heartbeat"] = time.ticks_ms() self.heartbeats[module_name]["status"] = "healthy" return True except Exception as e: print(f"[WDG] Module {module_name} recovery failed: {e}") self._log_fault(module_name, str(e)) return False def _all_critical_modules_recovered(self): """检查所有曾处于 critical 的模块是否已恢复""" for name, hb in self.heartbeats.items(): if hb["status"] != "healthy": return False return True def _log_fault(self, module, reason): """记录故障日志到内存和文件""" fault_info = { "time": time.time(), "module": module, "reason": reason } self.fault_log.append(fault_info) # 写入文件系统,便于下次启动后分析 try: log_path = "/data/wdt_faults.json" try: with open(log_path, "r") as f: logs = json.load(f) except (OSError, ValueError): logs = [] logs.append(fault_info) logs = logs[-50:] # 最多保留 50 条 with open(log_path, "w") as f: json.dump(logs, f) except Exception as e: print(f"[WDG] Failed to write fault log: {e}") def force_system_reset(self, reason="unknown"): """主动触发系统软复位""" print(f"[WDG] Forcing system reset, reason: {reason}") self._log_fault("system", reason) # 保存一些关键状态到 NVS 或文件 try: with open("/data/last_reset_reason.txt", "w") as f: f.write("{} - {}".format(time.time(), reason)) except Exception: pass machine.reset()5.2 调用示例:一个模拟多任务场景
# main.py import time import machine from wdt_system import SoftwareWatchdog # 创建看门狗实例 wdt = SoftwareWatchdog(hw_wdt_timeout_ms=15000) # 模拟模块1:传感器读取 def sensor_task_heartbeat(): # 实际代码里,这里调用传感器驱动读取数据 wdt.heartbeat("sensor") print("[Task] Sensor heartbeat") def sensor_recovery(): """传感器模块的恢复策略:重新初始化 I2C 并重新扫描设备""" print("[Recovery] Reinitializing sensor bus...") # 模拟重新初始化 time.sleep_ms(100) # 假设重新初始化成功 print("[Recovery] Sensor bus reinitialized, device found") return True # 模拟模块2:网络连接 def network_task_heartbeat(): wdt.heartbeat("network") print("[Task] Network heartbeat") def network_recovery(): """网络模块的恢复策略:重新连接 WiFi""" print("[Recovery] Reconnecting WiFi...") # 模拟断开重连 time.sleep_ms(200) print("[Recovery] WiFi reconnected") # 初始化 wdt.enable_hw_wdt() wdt.register("sensor", timeout_ms=3000, max_failures=3) wdt.register("network", timeout_ms=5000, max_failures=2) wdt.register_recovery("sensor", sensor_recovery) wdt.register_recovery("network", network_recovery) # 模拟主循环 counter = 0 last_print = time.ticks_ms() while True: # 此处模拟两个任务各自独立运行 if counter % 10 == 0: sensor_task_heartbeat() if counter % 15 == 0: network_task_heartbeat() # 周期性执行看门狗检查 if counter % 5 == 0: status = wdt.check_and_recover() if status == "critical": print("[Main] System critical, triggering reset...") wdt.force_system_reset("watchdog_critical") counter += 1 time.sleep_ms(100) # 模拟一个会导致 sensor 卡死的故障(在第 50 轮开始) if counter == 50: print("[Main] Simulating sensor hang... total stop sending heartbeat") # 不再调用 sensor_task_heartbeat()这段代码里有个很有意思的设计点:wdt.enable_hw_wdt()会开启硬件看门狗,而check_and_recover()方法里在检查前会先喂硬件看门狗。这样即使某个模块的恢复函数执行时间过长(比如网络重连卡住了),只要check_and_recover还能被主循环周期性调用,硬件看门狗就不会触发。而如果连主循环都卡死了,硬件看门狗最终会兜底复位。
5.3 升级版:接入 asyncio 的异步看门狗
如果你的项目用的是 MicroPython 的异步编程模型(asyncio),那软件看门狗可以做得更优雅。可以单独跑一个异步任务专门做监控,这样即使某个协程卡死了,监控任务依然在运行。
import uasyncio as asyncio import time class AsyncWatchdog: def __init__(self, interval_ms=1000): self.interval_ms = interval_ms self.heartbeats = {} self.failure_counts = {} self.thresholds = {} def register(self, name, timeout_ms, max_failures=3): self.heartbeats[name] = time.ticks_ms() self.failure_counts[name] = 0 self.thresholds[name] = (timeout_ms, max_failures) async def task_heartbeat(self, name, interval_ms=None): """让一个协程自己周期性汇报心跳""" interval = interval_ms or self.interval_ms while True: self.heartbeats[name] = time.ticks_ms() await asyncio.sleep_ms(interval) async def monitor_loop(self): """监控循环,持续检查所有注册模块的状态""" while True: await asyncio.sleep_ms(self.interval_ms) now = time.ticks_ms() for name, last_hb in self.heartbeats.items(): timeout_ms, max_failures = self.thresholds[name] elapsed = time.ticks_diff(now, last_hb) if elapsed > timeout_ms: print(f"[WDG] {name} category timeout: {elapsed}ms") self.failure_counts[name] += 1 if self.failure_counts[name] >= max_failures: print(f"[WDG] {name} critical, need recovery") # 在这里调用恢复逻辑 else: self.failure_counts[name] = 0这种写法在工程上非常干净。每个业务协程通过asyncio.create_task(wdt.task_heartbeat("sensor"))上报心跳,监控循环独立运行,互不干扰。当某个协程因为异常退出时,它的心跳也就自然停了,监控循环就能第一时间发现。
6. 实际部署中的关键细节与踩坑记录
代码能写出来是一回事,能在实际项目中稳定运行是另一回事。这节我把这段时间在 ESP32 和 RP2040 上部署软件看门狗遇到的几个典型问题整理出来,这些问题在官方文档里基本看不到。
6.1 喂狗频率和 timeout 的匹配问题
很多参考代码里直接把 WDT timeout 设置成 5 秒、喂狗间隔设置成 2 秒,看起来没问题,但实际运行时会因为time.ticks_ms()的精度、任务调度的不确定性导致偶发超时。我后来总结出一个经验公式:
喂狗周期最好小于等于 timeout 的 1/3。
也就是说 timeout 设 9 秒,喂狗周期最多 3 秒,留出三倍的裕量。这是因为 MicroPython 在 GC(垃圾回收)的时候会暂停所有 Python 代码执行,如果恰好在你准备调用feed()之前开始了一次较长的 GC(比如内存碎片严重时 GC 可能耗时数百毫秒甚至更久),喂狗会被推迟。三倍裕量基本能覆盖这种偶发情况。
另外要注意,MicroPython 的time.ticks_ms()是毫秒级计数器,但它的精度和稳定性取决于平台。在 ESP32 上实测精度还可以,在 RP2040 上就出现过 ticks 跳变的情况,所以不要对超时判断做过于严格的边界值处理,至少要留出 10% 的容差。
6.2 恢复函数执行期间的“假死”问题
设计恢复函数时,最容易忽略的一点是:恢复函数本身也可能卡死。比如网络重连函数里用了socket.connect()而没有设置超时,如果网络环境特别差,这个调用可能阻塞几十秒。如果这段时间内所有模块都处于异常状态,等于整个系统卡在了恢复函数里,没有任何心跳,硬件看门狗会触发复位——这倒也不算是坏事,因为硬件看门狗就是兜底的,但问题是你失去了“分级恢复”的意义,软件看门狗退化成了一个普通的重启器。
解决思路有两个。第一个是在恢复函数里尽量使用带超时的调用,比如 socket 操作设置timeout参数。第二个是给恢复函数本身设置执行时限,用定时器机制检测恢复函数是否卡住,如果超时就立刻放弃恢复,直接走硬件复位流程。
我后来开发了一个简单的“恢复看门狗”机制:
class RecoveryTimeoutError(Exception): pass def execute_with_timeout(fn, timeout_ms, *args, **kwargs): """在指定时间内执行函数,超时抛异常""" result = [None] error = [None] done = [False] def wrapper(): try: result[0] = fn(*args, **kwargs) except Exception as e: error[0] = e finally: done[0] = True import _thread t = _thread.start_new_thread(wrapper, ()) start = time.ticks_ms() while not done[0]: if time.ticks_diff(time.ticks_ms(), start) > timeout_ms: raise RecoveryTimeoutError(f"Recovery function timeout after {timeout_ms}ms") time.sleep_ms(10) if error[0]: raise error[0] return result[0]但这里要提醒一下,_thread在 MicroPython 上并不是所有平台都可用,而且线程模式下对某些外设的访问会有竞争风险。因此我在实际项目中更多是采用“分段恢复”的策略:把恢复动作拆成多个小步骤,每步之间通过check_and_recover检查是否超时,而不是给恢复函数整体设时限。这样处理更安全。
6.3 循环里多个模块恢复时的执行顺序
如果你的看门狗同时监控了 5 个模块,在某次检查中发现 3 个模块都超时了,怎么决定恢复顺序?最开始我按字典遍历顺序来,结果发现这种顺序不太合理。后来我引入了“优先级”概念,每个模块注册时可以设置恢复优先级:
def register(self, name, timeout_ms=5000, max_failures=3, priority=10): # priority 数值越小优先级越高 self.thresholds[name] = (timeout_ms, max_failures, priority)恢复的时候按优先级从高到低排序执行。比如网络模块优先级比传感器模块高,因为传感器数据要上报必须先有网络连接,先恢复网络再恢复传感器,整体恢复效率会高很多。
不过还有一个微妙的地方,有些模块的恢复是有依赖关系的,单纯的优先级排序解决不了。比如网络模块的恢复需要读取配置文件,而配置文件模块本身也在超时列表中。这种场景我建议不要试图在软件看门狗里解决,而是把恢复逻辑收敛到更上层的“系统软复位”里,一次性重新初始化所有相关模块。
6.4 故障日志写入 flash 的坑
记录故障日志这个需求听起来很简单,写文件嘛。但在真实设备上这里有个隐藏问题:flash 的擦写寿命是有限的。ESP32 的 flash 虽然有磨损均衡,但如果你每次看门狗触发都写一次日志,写得太频繁也会加速 flash 老化。
我的做法有两个:一是在内存中维护一个环形缓冲区,只有缓冲满了或者系统即将重启时才把完整日志写入 flash;二是在写入时限制日志文件大小,只保留最近几十条。
class FaultLogBuffer: def __init__(self, capacity=10): self.buffer = [] self.capacity = capacity def append(self, entry): self.buffer.append(entry) if len(self.buffer) > self.capacity: # 只保留最近 capacity 条 self.buffer = self.buffer[-self.capacity:] def flush_to_file(self, filepath="/data/wdt_faults.json"): try: existing = [] try: with open(filepath, "r") as f: existing = json.load(f) except Exception: existing = [] merged = existing + self.buffer merged = merged[-self.capacity:] with open(filepath, "w") as f: json.dump(merged, f) self.buffer = [] except Exception as e: print(f"[LogBuffer] Failed to flush: {e}")6.5 标定“正常运行”的心跳周期
最后一个问题是关于“心跳周期”的设定。很多模块的更新频率并不是恒定的,比如传感器可能在启动阶段需要 8 秒才能完成初始化,但正常运行后每 2 秒就能读一次。如果按正常频率设置心跳超时,启动阶段就会被误判为故障。
解决方式是多档阈值设计。我实现了一个update_threshold方法,允许在系统启动、运行、休眠等不同阶段动态调整心跳阈值:
def update_threshold(self, name, timeout_ms, max_failures=None): """动态调整模块心跳阈值""" if name in self.thresholds: _, old_max_failures, priority = self.thresholds[name] new_max_failures = max_failures if max_failures is not None else old_max_failures self.thresholds[name] = (timeout_ms, new_max_failures, priority) print(f"[WDG] Updated {name} threshold to {timeout_ms}ms / {new_max_failures}failures")比如在启动阶段,传感器模块调用update_threshold("sensor", timeout_ms=10000),等初始化完成后再改回timeout_ms=3000。这样就避免了启动时的误报。
7. 进阶思考:把软件看门狗从“应急措施”变成“诊断工具”
文章写到这里,可能有人会觉得软件看门狗本质就是一个“兜底程序”,只要触发就说明系统已经出问题了,所以它的价值主要在恢复。但我在实际项目中体会最深的一点是:看门狗最宝贵的数据不是你写了多少行恢复代码,而是你通过看门狗收集到的那些异常信息。
7.1 用故障反推系统短板
我第一次完整部署这套看门狗系统后,跑了一周,日志文件里记录了三次故障,全部来自传感器模块。仔细分析日志后发现,触发时间基本集中在凌晨三点左右。后来排查发现,那个时间点附近的蓝牙广播干扰导致 I2C 总线持续忙。知道了这个规律之后,我在传感器驱动里加了总线恢复机制,从此传感器模块再也没出现过超时。
这个经验告诉我们:看门狗日志不是拿来“存档”的,而是拿来“破案”的。设计时就应该在日志里输出尽可能多的上下文信息,比如当前各模块状态、内存剩余量、CPU 负载、最近一次喂狗时间等。这样才能在故障发生后快速定位根因。
7.2 与云平台配合实现远程诊断
如果设备具备网络能力,我强烈建议把看门狗的告警和日志同步上云。最简单的做法是在触发恢复时发送一条 MQTT 消息到云端,内容包括模块名称、故障原因、内存使用率等。这样即使设备不在手边,也能第一时间看到异常。
def send_fault_alert(module, reason, mem_free): import ujson from umqtt.simple import MQTTClient client = MQTTClient("watchdog_alert", "192.168.1.100", port=1883) client.connect() payload = ujson.dumps({ "device_id": "esp32_s3_01", "module": module, "reason": reason, "mem_free": mem_free, "timestamp": time.time() }) client.publish("iot/watchdog_alert", payload) client.disconnect()当然,这里有个悖论:如果网络模块本身就出了问题,MQTT 消息大概率发不出去。所以这个远程告警更适合作为辅助手段,不能完全依赖。
7.3 从“复位”到“自愈”的演进方向
我认为做嵌入式软件开发,最终追求的状态不是“出问题了能快速重启”,而是“系统自己具备免疫力”。软件看门狗只是建立免疫力的第一步。后续可以沿着这几个方向继续演进:
- 基于故障日志的根因分析,不断优化业务逻辑,从源头减少故障发生的概率。
- 引入配置热更新机制,让设备在出现异常后能自动切换到降级配置运行,而不是死守完整功能。
- 设计异常自测流程,在系统空闲时定期做一次“自检”,提前发现可能出问题的硬件或外设。
这些方向展开讲又是好几篇文章的篇幅,但在设计看门狗时如果时刻带着“自愈”这个目标,整体架构会比你想象的好很多。回头看我自己第一次写的看门狗代码,其实就是个“定时重启器”,但现在这套系统已经能智能分辨出哪根线松了、哪个总线卡了、哪些场景只需要局部恢复——这才是软件看门狗真正的价值所在。