去年在做一块 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(timeout5000) # 单位是毫秒 # 主循环中周期性喂狗 while True: # ... 业务逻辑 ... wdt.feed()注意MicroPython 的machine.WDT实现的是硬件看门狗功能。也就是说你一旦创建了 WDT 实例喂狗周期就必须小于 timeout否则系统会被强制复位。这是硬件行为不是 Python 层的软逻辑。3.1 network 场景下 WDT 的误触发现象用machine.WDT的过程中我踩过一个大坑如果你的主循环里用了阻塞式的网络请求比如requests.get()而且没有设置超时那么当网络异常时这个请求可能阻塞几十秒甚至更久硬件看门狗就会因为没被喂到而触发复位。你可能会说网络请求不是有 timeout 参数吗确实有但有些库在底层 TCP 重传机制里timeout 并不可靠。我遇到过设置了 timeout10 但实际阻塞了 45 秒的情况。所以如果你在带网络功能的项目里用machine.WDT一定要确保喂狗操作发生在主循环最外层并且在所有可能长时间阻塞的调用前后都至少要保证主循环有返回的机会。一个比较稳妥的做法是把喂狗操作放在定时器中断里from machine import Timer, WDT wdt WDT(timeout5000) def feed_wdt(timer): wdt.feed() # 每 2 秒喂一次 timer Timer(0) timer.init(period2000, modeTimer.PERIODIC, callbackfeed_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_ms5000, max_failures3): 注册一个需要监控的模块 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, sclPin(22), sdaPin(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_ms10000): 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(timeoutself.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_ms5000, max_failures3): 注册监控模块 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), ffailure 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, reasonunknown): 主动触发系统软复位 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_ms15000) # 模拟模块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_ms3000, max_failures3) wdt.register(network, timeout_ms5000, max_failures2) 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_ms1000): self.interval_ms interval_ms self.heartbeats {} self.failure_counts {} self.thresholds {} def register(self, name, timeout_ms, max_failures3): 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_msNone): 让一个协程自己周期性汇报心跳 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(fRecovery 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_ms5000, max_failures3, priority10): # priority 数值越小优先级越高 self.thresholds[name] (timeout_ms, max_failures, priority)恢复的时候按优先级从高到低排序执行。比如网络模块优先级比传感器模块高因为传感器数据要上报必须先有网络连接先恢复网络再恢复传感器整体恢复效率会高很多。不过还有一个微妙的地方有些模块的恢复是有依赖关系的单纯的优先级排序解决不了。比如网络模块的恢复需要读取配置文件而配置文件模块本身也在超时列表中。这种场景我建议不要试图在软件看门狗里解决而是把恢复逻辑收敛到更上层的“系统软复位”里一次性重新初始化所有相关模块。6.4 故障日志写入 flash 的坑记录故障日志这个需求听起来很简单写文件嘛。但在真实设备上这里有个隐藏问题flash 的擦写寿命是有限的。ESP32 的 flash 虽然有磨损均衡但如果你每次看门狗触发都写一次日志写得太频繁也会加速 flash 老化。我的做法有两个一是在内存中维护一个环形缓冲区只有缓冲满了或者系统即将重启时才把完整日志写入 flash二是在写入时限制日志文件大小只保留最近几十条。class FaultLogBuffer: def __init__(self, capacity10): 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_failuresNone): 动态调整模块心跳阈值 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_ms10000)等初始化完成后再改回timeout_ms3000。这样就避免了启动时的误报。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, port1883) 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 从“复位”到“自愈”的演进方向我认为做嵌入式软件开发最终追求的状态不是“出问题了能快速重启”而是“系统自己具备免疫力”。软件看门狗只是建立免疫力的第一步。后续可以沿着这几个方向继续演进基于故障日志的根因分析不断优化业务逻辑从源头减少故障发生的概率。引入配置热更新机制让设备在出现异常后能自动切换到降级配置运行而不是死守完整功能。设计异常自测流程在系统空闲时定期做一次“自检”提前发现可能出问题的硬件或外设。这些方向展开讲又是好几篇文章的篇幅但在设计看门狗时如果时刻带着“自愈”这个目标整体架构会比你想象的好很多。回头看我自己第一次写的看门狗代码其实就是个“定时重启器”但现在这套系统已经能智能分辨出哪根线松了、哪个总线卡了、哪些场景只需要局部恢复——这才是软件看门狗真正的价值所在。 MicroPython软件看门狗实战：从死循环到自愈的嵌入式恢复机制
去年在做一块 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(timeout5000) # 单位是毫秒 # 主循环中周期性喂狗 while True: # ... 业务逻辑 ... wdt.feed()注意MicroPython 的machine.WDT实现的是硬件看门狗功能。也就是说你一旦创建了 WDT 实例喂狗周期就必须小于 timeout否则系统会被强制复位。这是硬件行为不是 Python 层的软逻辑。3.1 network 场景下 WDT 的误触发现象用machine.WDT的过程中我踩过一个大坑如果你的主循环里用了阻塞式的网络请求比如requests.get()而且没有设置超时那么当网络异常时这个请求可能阻塞几十秒甚至更久硬件看门狗就会因为没被喂到而触发复位。你可能会说网络请求不是有 timeout 参数吗确实有但有些库在底层 TCP 重传机制里timeout 并不可靠。我遇到过设置了 timeout10 但实际阻塞了 45 秒的情况。所以如果你在带网络功能的项目里用machine.WDT一定要确保喂狗操作发生在主循环最外层并且在所有可能长时间阻塞的调用前后都至少要保证主循环有返回的机会。一个比较稳妥的做法是把喂狗操作放在定时器中断里from machine import Timer, WDT wdt WDT(timeout5000) def feed_wdt(timer): wdt.feed() # 每 2 秒喂一次 timer Timer(0) timer.init(period2000, modeTimer.PERIODIC, callbackfeed_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_ms5000, max_failures3): 注册一个需要监控的模块 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, sclPin(22), sdaPin(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_ms10000): 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(timeoutself.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_ms5000, max_failures3): 注册监控模块 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), ffailure 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, reasonunknown): 主动触发系统软复位 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_ms15000) # 模拟模块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_ms3000, max_failures3) wdt.register(network, timeout_ms5000, max_failures2) 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_ms1000): self.interval_ms interval_ms self.heartbeats {} self.failure_counts {} self.thresholds {} def register(self, name, timeout_ms, max_failures3): 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_msNone): 让一个协程自己周期性汇报心跳 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(fRecovery 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_ms5000, max_failures3, priority10): # priority 数值越小优先级越高 self.thresholds[name] (timeout_ms, max_failures, priority)恢复的时候按优先级从高到低排序执行。比如网络模块优先级比传感器模块高因为传感器数据要上报必须先有网络连接先恢复网络再恢复传感器整体恢复效率会高很多。不过还有一个微妙的地方有些模块的恢复是有依赖关系的单纯的优先级排序解决不了。比如网络模块的恢复需要读取配置文件而配置文件模块本身也在超时列表中。这种场景我建议不要试图在软件看门狗里解决而是把恢复逻辑收敛到更上层的“系统软复位”里一次性重新初始化所有相关模块。6.4 故障日志写入 flash 的坑记录故障日志这个需求听起来很简单写文件嘛。但在真实设备上这里有个隐藏问题flash 的擦写寿命是有限的。ESP32 的 flash 虽然有磨损均衡但如果你每次看门狗触发都写一次日志写得太频繁也会加速 flash 老化。我的做法有两个一是在内存中维护一个环形缓冲区只有缓冲满了或者系统即将重启时才把完整日志写入 flash二是在写入时限制日志文件大小只保留最近几十条。class FaultLogBuffer: def __init__(self, capacity10): 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_failuresNone): 动态调整模块心跳阈值 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_ms10000)等初始化完成后再改回timeout_ms3000。这样就避免了启动时的误报。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, port1883) 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 从“复位”到“自愈”的演进方向我认为做嵌入式软件开发最终追求的状态不是“出问题了能快速重启”而是“系统自己具备免疫力”。软件看门狗只是建立免疫力的第一步。后续可以沿着这几个方向继续演进基于故障日志的根因分析不断优化业务逻辑从源头减少故障发生的概率。引入配置热更新机制让设备在出现异常后能自动切换到降级配置运行而不是死守完整功能。设计异常自测流程在系统空闲时定期做一次“自检”提前发现可能出问题的硬件或外设。这些方向展开讲又是好几篇文章的篇幅但在设计看门狗时如果时刻带着“自愈”这个目标整体架构会比你想象的好很多。回头看我自己第一次写的看门狗代码其实就是个“定时重启器”但现在这套系统已经能智能分辨出哪根线松了、哪个总线卡了、哪些场景只需要局部恢复——这才是软件看门狗真正的价值所在。