做嵌入式开发这些年,我遇到过最头疼的问题不是“设备上电没反应”,而是“设备跑着跑着就沉默了”。尤其是用 MicroPython 写业务逻辑的项目,开发效率确实高,但微控制器资源有限、任务一多、状态一复杂,某个模块悄悄卡死是常有的事。更麻烦的是,就算加了硬件看门狗,也未必能兜得住——因为很多时候程序并没有整体崩溃,主循环还在正常运行,某个任务却已经失去了响应。这种业务级卡死,恰好需要软件看门狗来盯梢,再配合恢复机制让设备自己把自己拉回来。这篇文章就手把手带你在 MicroPython 里实现一个带分级恢复机制的软件看门狗,从设计思路、完整代码到常见坑位,一次性讲透。
1. 为什么 MicroPython 项目需要软件看门狗
1.1 硬件看门狗的盲区:系统活着不代表任务活着
先聊一个很实际的场景。我做过一个环境监测网关,MicroPython 跑在 ESP32 上,核心业务是数据采集、屏幕刷新、网络上报三个模块。硬件看门狗我一开始就加了,按说系统崩溃会自动复位。结果设备运行到第三天,网络模块不再发包,屏幕停留在最后一张界面,指示灯也不闪了——但主循环还在跑,喂狗动作一直正常,硬件看门狗压根没被触发。
这就是硬件看门狗最典型的盲区。它只能判断“主循环有没有在喂”,判断不了“每个业务任务是否真的在工作”。很多故障根本不是整个系统挂死,而是某个任务陷入了逻辑死锁:等待一个永远不会到来的事件、在某个状态机分支里空转、或者拿到了不该拿到的东西不释放。主循环照常调度,喂狗照常执行,硬件看门狗永远给你报平安,业务却已经停滞了。
MicroPython 项目这种情况更明显,因为 MicroPython 通常跑的是业务逻辑密集的脚本,模块之间缺乏完善的异常隔离,一个任务出问题很容易“带病运行”。加了硬件看门狗只是最后一道防线,它没有业务视角,指望它解决业务级故障不现实。软件看门狗的作用,就是补上这一层:它专门盯着各个任务的心跳,谁不干活了就把它揪出来处理。
1.2 软件看门狗要解决的核心问题
软件看门狗本质上解决的是“任务级故障检测与恢复”。它要做的事情只有三件:知道任务应该多久报告一次心跳、持续检查任务有没有按时报告、发现超时后执行对应的恢复动作。
什么叫心跳?你可以把任务想象成员工,看门狗是监工。每个员工干活时要定期打卡,监工每隔一段时间看一眼打卡记录。哪个员工超过规定时间没打卡,就说明他可能出问题了,监工需要介入。
这里的关键是“任务级”三个字。软件看门狗关注的不是整个系统是否在跑,而是每个具体任务是否还在正常推进。数据采集任务要定期采集数据,显示任务要定期刷新屏幕,通信任务要定期处理收发请求。每个任务自带一个最后心跳时间戳,看门狗周期巡检时逐个比对当前时间与时间戳,超过阈值就判定故障。
与硬件看门狗相比,软件看门狗最大的优势是恢复动作可以做到精细化。硬件看门狗超时只能系统级复位,软件看门狗可以对单个任务做重置、重启,甚至直接软复位整个系统,可以分等级处理,而不是一刀切。
1.3 恢复机制不是“重启了之”
很多初学者把软件看门狗做成了“超时即重启”,这其实是把硬件看门狗搬到软件里,白瞎了恢复机制的潜力。恢复机制的正确姿势,是让系统具备“自愈能力”,用最小的代价恢复业务。
举一个工厂里的类比:一台机器某个传感器数据异常,最粗暴的修法是整机断电重启,生产全停。合理修法是先重启传感器模块,不行再重启控制程序,还不行才考虑整机下电。恢复机制的思路就是这样的:一级恢复清理任务状态,二级恢复重启任务,三级恢复软复位系统。这样大部分偶发故障在低等级就处理掉了,根本不影响其他任务。
另一个容易被忽略的点是“故障计数”。单次超时不一定意味着任务真的坏了,可能是某次调度被阻塞了一下。如果一超时就恢复,系统会非常敏感,任务执行周期稍有波动就误报。所以通常采用连续超时多次判定,比如连续两次检测超时再触发恢复,排除偶发抖动。
2. 整体设计思路与方案选型
2.1 核心机制:心跳上报 + 周期巡检
软件看门狗的整体架构不复杂,核心是两个角色:被监控的任务和看门狗管理器。被监控的任务在每个运行周期内主动调用feed()上报心跳,更新自己的最后心跳时间戳;看门狗管理器用定时器周期巡检所有任务,判断每个任务的最后心跳时间是否已经超过timeout_ms。
巡检判定有一个细节很重要:时间戳比较要用time.ticks_ms()和time.ticks_diff(),而不是直接相减。MicroPython 的毫秒计数器是有限位数的,大约 49 天后会发生回绕,直接做减法在回绕瞬间会算出个极大的负数,导致所有任务同时被误判为超时。用time.ticks_diff(now, last_alive_ms)则是 MicroPython 官方推荐的方式,底层已经处理了回绕问题。这个坑我当初踩过一次,排查了半天,最后发现是时间回绕把看门狗搞疯了。
心跳上报必须由任务自身的运行路径发起,不能由其他任务代劳,这点下文会展开。正常运行时,任务每周期调一次feed(),更新时间戳并清零连续故障计数。一旦任务逻辑卡死,它就不会再调用feed(),时间戳停留在卡死前的时刻,看门狗就能立刻察觉到异常。
2.2 为什么用定时器而不是多线程
实现周期性巡检有两种常见方案:开一个线程循环检查,或者用machine.Timer定时回调。MicroPython 项目里我更推荐定时器方案,原因有三。
第一,MicroPython 的_thread线程支持在不同硬件平台上表现参差不齐。大多数 MicroPython 移植版没有真正的多核并行,线程调度依赖分片轮转,实时性没有保证。线程还容易和底层外设中断产生竞争,调试起来非常痛苦。用定时器回调则简单得多,平台一致性也更好。
第二,定时器的语义更贴合“周期巡检”这件事。machine.Timer初始化时指定周期,到达周期后自动触发回调,代码结构天然就是“每隔固定时间检查一次”,不容易出现线程里漏循环、循环被阻塞的问题。
第三,线程方案还需要考虑锁、队列、同步状态,代码量成倍增加。看门狗本身是个系统基础组件,越简单越不容易出错。定时器回调 + 主循环消费恢复事件的模式,是目前我见过的最可靠、也最容易维护的软件看门狗结构。
2.3 三级恢复策略:从轻到重、逐级升级
我设计恢复机制时参考了操作系统的容错理念:故障恢复应该“尽量局部化、尽量低成本”。因此把恢复动作分成三个等级,等级越高影响面越大。
一级恢复调用任务的reset()方法。适用于任务内部状态错乱但代码本体还能运行的情况,比如数据采集任务里计数器失控、缓存区数据一致性破坏。reset()负责把任务内部状态恢复到初始值,不中断任务继续运行,对系统的影响最小。
二级恢复调用任务的restart()方法。适用于任务状态已经无法通过简单重置恢复的情况,比如任务内部出现异常分支,运行逻辑走入死胡同。restart()会停止任务当前活动,重新执行初始化流程,相当于单个任务“冷启动”。
三级恢复执行machine.reset()软复位整个系统。适用于前两等级恢复后依然反复出问题的场景,说明故障根源可能已经影响全局,或者任务涉及的内存、外设状态无法在应用层恢复。软复位虽然代价大,但保证系统不至于彻底瘫痪。
恢复等级不是固定不变的,而是采用“逐级升级”策略。任务第一次触发恢复时用一级,很快再次触发就用二级,再不行就三级。同时,如果任务恢复正常运行足够久,恢复等级会逐步回落,避免一次历史故障导致以后永远走最高等级恢复。
2.4 关键参数怎么定才有实操参考价值
软件看门狗的稳定性很大程度上取决于参数设置,尤其是timeout_ms。我实际调试中总结了一套经验值,可以直接参考。
巡检周期check_period_ms一般取 100ms 到 500ms。太短会增加定时器中断频率,消耗 CPU;太长会导致故障发现不及时。我的经验是取任务正常上报周期的中间值,比如任务每 50ms 喂一次狗,巡检设 200ms,这样最坏情况 250ms 内能发现故障,足够灵敏又不会过于频繁。
超时阈值timeout_ms我通常设置为任务正常运行周期的 2 到 5 倍。比如数据采集任务每 500ms 执行一次,超时阈值就设在 1000ms 到 2500ms 之间。设置过小容易在任务调度抖动时误报,设置过大会让故障发现变得迟钝。这里的核心思想是:给任务留足正常运行时的余量,但绝不能让它无限拖延。
连续故障阈值max_faults我一般取 2。单次超时可能只是偶发抖动,连续两次超时基本能确定任务真的出了问题。如果项目对可靠性要求极严,可以取 3,再多就没有意义了,反而延误恢复时机。
3. 核心代码实现与关键细节解析
3.1 SoftWatchdog 管理器:注册、喂狗、巡检三位一体
看门狗管理器我封装成一个类SoftWatchdog,对外只暴露四个接口:register()注册任务、feed()上报心跳、start()启动巡检、process_recovery_events()执行恢复动作。整体结构如下,可以直接复制到一个soft_watchdog.py文件里使用。
from machine import Timer import time class SoftWatchdog: def __init__(self, check_period_ms=200): self.check_period_ms = check_period_ms self.tasks = {} self.timer = Timer(0) self._in_isr = False def register(self, name, instance, timeout_ms=5000, max_faults=2): self.tasks[name] = { 'instance': instance, 'timeout_ms': timeout_ms, 'max_faults': max_faults, 'last_alive_ms': time.ticks_ms(), 'fault_count': 0, 'recovery_count': 0, 'healthy_count': 0, 'recovery_pending': False, 'pending_level': 0, } def feed(self, name): task = self.tasks.get(name) if task is None: return task['last_alive_ms'] = time.ticks_ms() task['fault_count'] = 0 def start(self): self.timer.init(period=self.check_period_ms, mode=Timer.PERIODIC, callback=self._on_tick) def stop(self): self.timer.deinit() def process_recovery_events(self): for name, task in self.tasks.items(): if not task['recovery_pending']: continue task['recovery_pending'] = False level = task['pending_level'] task['pending_level'] = 0 self._execute_recovery(name, task['instance'], level) def _on_tick(self, timer): if self._in_isr: return self._in_isr = True try: self._check_all() finally: self._in_isr = False def _check_all(self): now = time.ticks_ms() for name, task in self.tasks.items(): if task['recovery_pending']: continue elapsed = time.ticks_diff(now, task['last_alive_ms']) if elapsed <= task['timeout_ms']: task['healthy_count'] += 1 if task['healthy_count'] >= 50: task['healthy_count'] = 0 if task['recovery_count'] > 0: task['recovery_count'] -= 1 continue task['healthy_count'] = 0 task['fault_count'] += 1 if task['fault_count'] < task['max_faults']: continue task['fault_count'] = 0 task['recovery_count'] += 1 task['pending_level'] = self._decide_level(task['recovery_count']) task['recovery_pending'] = True @staticmethod def _decide_level(recovery_count): if recovery_count <= 1: return 1 if recovery_count == 2: return 2 return 3 def _execute_recovery(self, name, instance, level): print('[WDT] recover task %s level %d' % (name, level)) try: if level == 1: if hasattr(instance, 'reset'): instance.reset() elif level == 2: if hasattr(instance, 'restart'): instance.restart() else: import machine machine.reset() except Exception as e: print('[WDT] recovery fail: %s' % e)每个任务在register()时传入一个实例对象,这个对象需要实现reset()和restart()方法。如果一个任务连reset()都没有,一级恢复就没有动作可做,实际项目中这本身就是个设计问题,说明任务对象没有考虑恢复能力。代码里hasattr()判断可以兜底,但更靠谱的做法是每个任务都主动实现这两个方法。
3.2 巡检回调:中断上下文里只做轻量操作
_on_tick是定时器回调,在多数 MicroPython 移植版上运行在中断上下文或者非常受限的环境中,这决定了它只能做轻量操作。我在里面加了一个_in_isr标志做重入保护,防止回调重入导致数据错乱。这是很多初版看门狗最容易忽略的地方。
在_check_all()里,巡检逻辑只做几件事:计算时间差、维护计数器、修改标志位。这些操作都是纯数据运算,不涉及内存分配,不调用打印,不执行 I/O,安全性有保障。尤其要注意,某些 MicroPython 平台在中断上下文里执行print()或创建新对象会直接抛异常甚至导致系统崩溃,所以巡检路径上坚决不做这些事。
那恢复动作放哪执行?答案在主循环。process_recovery_events()必须由主循环周期调用,它的作用是扫描所有任务的recovery_pending标志,发现标志置位才去执行reset()、restart()或machine.reset()。这样设计的好处是:恢复动作虽然可能耗时较长,但不在中断上下文执行,不会影响系统定时器回调的稳定性;同时主循环天然串行处理恢复事件,不需要加锁。
另外一个细节是故障超时的判定逻辑:我先检查elapsed是否超过timeout_ms,超时后fault_count加一,但只有fault_count达到max_faults才真正触发恢复。这样单次偶发超时不会引起恢复动作,只有连续多次超时才判定任务确实出问题,大幅度降低误报率。
3.3 恢复等级的动态升降:故障计数和健康计数配合
恢复等级的设计结合了两个计数器:recovery_count和healthy_count。recovery_count记录任务累计触发恢复的次数,触发一次加一,等级由它决定。这样如果任务反复出问题,恢复动作会逐级升级,直至软复位整个系统,体现出“从轻到重、逐级升级”的容错思想。
healthy_count则用来做恢复等级的回退。任务每次正常巡检通过后healthy_count加一,达到设定阈值(代码里是 50 次)就说明任务已经稳定运行了一段时间,此时把recovery_count减一。也就是说,如果任务在恢复后正常运行了几十秒,恢复等级会自然降低,下次故障再从低等级开始尝试。这个机制避免了任务恢复正常后还被“历史包袱”压着的问题,让恢复策略更加智能。
这两个计数值在代码里是完全独立的,互不干扰。fault_count负责单次故障的连续计数,recovery_count负责历史故障的累计与等级判断,healthy_count负责等级衰减。三者配合,整个恢复机制既有灵敏度又有稳定性。
4. 完整演示:让一个卡死的任务自动自愈
4.1 环境准备与工程结构
演示代码我以 ESP32-S3 开发板为例,MicroPython 固件版本 1.20 以上即可,大部分特性只要是支持machine.Timer的平台都能跑。工程只需要两个文件:soft_watchdog.py存放看门狗管理器,main.py存放业务任务和主循环。用 Thonny 编辑器连上开发板,把两个文件传上去就能运行。
把soft_watchdog.py和main.py分别传到根目录后,开发板重新上电或软复位就会自动执行main.py。如果你手头是 STM32、RP2040 这类也支持 MicroPython 的板子,代码几乎不用改动,只要确认板子支持machine.Timer即可。这三个平台我都实测过,machine.Timer的行为基本一致。
4.2 演示任务与故障注入:模拟业务逻辑死锁
为了让演示贴近真实问题,我设计了两个任务:Collector模拟数据采集任务,正常时每次执行counter加一;DisplayTask模拟显示刷新任务,正常时刷新frames计数。关键点在Collector里加了一个hung标志位,一旦置位,run_once()直接返回,模拟业务逻辑死锁——任务既不干活,也不再主动上报心跳。
这里要特别强调喂狗位置的设计。主循环里喂狗时写了判断:if not collector.hung: wdt.feed('collector')。也就是说,只有Collector正常执行时才喂狗,一旦进入hung状态,它自己的心跳就断了。千万不要在主循环开头无条件喂所有任务,那是把看门狗当摆设。
import time from soft_watchdog import SoftWatchdog WDT_CHECK_PERIOD_MS = 200 COLLECTOR_TIMEOUT_MS = 2000 DISPLAY_TIMEOUT_MS = 1000 class Collector: def __init__(self): self.counter = 0 self.hung = False def run_once(self): if self.hung: # 模拟业务逻辑死锁:不干活,也不再上报心跳 return self.counter += 1 def set_hung(self, hung): self.hung = hung def reset(self): print('[Collector] reset() called') self.hung = False self.counter = 0 def restart(self): print('[Collector] restart() called') self.hung = False self.counter = 0 class DisplayTask: def __init__(self): self.frames = 0 def update(self): self.frames += 1 def reset(self): print('[Display] reset() called') self.frames = 0 def main(): collector = Collector() display = DisplayTask() wdt = SoftWatchdog(check_period_ms=WDT_CHECK_PERIOD_MS) wdt.register('collector', collector, timeout_ms=COLLECTOR_TIMEOUT_MS, max_faults=2) wdt.register('display', display, timeout_ms=DISPLAY_TIMEOUT_MS, max_faults=2) wdt.start() hang_plan = {100: True, 500: True, 900: True} tick = 0 while True: wdt.process_recovery_events() collector.run_once() display.update() if not collector.hung: wdt.feed('collector') wdt.feed('display') if tick in hang_plan: print('simulate collector hang at tick %d' % tick) collector.set_hung(True) tick += 1 time.sleep_ms(50) if __name__ == '__main__': main()故障注入方案也很简单:hang_plan里定义了几个时间点,分别在 tick 等于 100、500、900 时把Collector置为挂起。这样可以在一个运行周期内连续测试三次故障恢复过程,验证恢复等级是否逐级升级。
4.3 实测日志与恢复过程分析
在 ESP32-S3 上运行这段代码,串口输出大致如下:
simulate collector hang at tick 100 [WDT] recover task collector level 1 [Collector] reset() called simulate collector hang at tick 500 [WDT] recover task collector level 2 [Collector] restart() called simulate collector hang at tick 900 [WDT] recover task collector level 3 [WDT] machine reset now第一次挂起发生在 tick 100,也就是系统启动约 5 秒后。Collector停止喂狗,看门狗巡检发现超时,连续两次确认故障后触发一级恢复,调用reset(),输出日志可以看到Collector被重置,随后恢复正常运行。
第二次挂起发生在 tick 500,此时recovery_count已经累加到 2,看门狗自动升级到二级恢复,调用restart()。日志里能明显看到第二次恢复动作与第一次不同。第三次挂起发生在 tick 900,这时recovery_count达到 3,看门狗直接执行三级恢复machine.reset(),开发板软复位。
这个输出清晰地说明了恢复机制的完整闭环:发现故障、判定故障、分级处理、逐级升级。而且整个过程中DisplayTask一直正常运行,看门狗只盯住了异常任务,没有影响其他任务的稳定性。这正是软件看门狗相比硬件看门狗的核心价值——处置精确、影响可控。
5. 常见问题与排查技巧实录
5.1 看门狗频繁误触发怎么排查
如果看门狗经常误报,第一个要查的是timeout_ms是否设置合理。任务正常运行周期可能不是固定值,比如网络通信任务在重连时可能阻塞几百毫秒,数据采集遇到传感器响应慢也可能拖长。这类波动如果超过timeout_ms,就会被看门狗误判为故障。
我的排查套路是:先在任务正常运行时不加看门狗,用逻辑分析仪或者直接在代码里记录每次喂狗的实际间隔,统计出最大周期。然后把这个最大周期乘以 2 到 3 作为timeout_ms。如果误报还是频繁,再检查是不是喂狗位置放错了,比如放在了try块内但某个异常分支把它跳过,或者放在了耗时操作之前而不是之后。
还有一种隐蔽的误报原因:任务喂狗的代码路径依赖了外设中断或回调,而这些中断回调本身被其他逻辑阻塞了。这种情况下喂狗间隔会受到系统整体调度影响,需要重新设计喂狗触发点,把它放到纯粹由任务自身状态驱动的路径上,不要依赖任何可能被阻塞的外设事件。
5.2 任务真的卡死了,但看门狗没反应
这个问题通常出在“喂狗的人和干活的人不是同一个”。最常见的情况是在主循环开头统一喂一遍所有任务,结果某个任务卡死后,主循环还在正常跑,喂狗动作由主循环代劳了,心跳就一直被续上,看门狗永远发现不了问题。
正确的喂狗姿势是:谁干活谁喂狗。每个任务必须在自己的run()方法、状态机流转或者处理函数内部主动调用feed()。只有任务自身的代码执行到了某个关键节点,才允许它向看门狗证明“我还活着”。如果任务卡死在业务逻辑里,它连走到喂狗这一行的机会都没有,看门狗才能可靠地探测到异常。
另外要注意,软件看门狗本身也有盲区——如果整个主循环都死掉了,比如某个任务里有个while True: pass,定时器回调虽然能发现超时、置位恢复标志,但主循环不再调用process_recovery_events(),恢复动作永远不会执行。这种情况只能靠硬件看门狗兜底。这也是我一直强调的:软件看门狗和硬件看门狗是互补关系,不是替代关系。实战项目中我通常会两者并用,软件负责任务级精准恢复,硬件负责系统级最后防线。
5.3 MicroPython 定时器回调的硬性约束
我最初写 SoftWatchdog 第一版时,把恢复动作直接放进了定时器回调里,结果在 ESP32 上运行了几次就出现随机崩溃。查了很久才确认是中断上下文里做了内存分配和打印操作触发了MemoryError,最终导致系统异常。
MicroPython 的定时器回调在多数移植版上运行在受限的上下文里,有几条禁令必须遵守:不能动态分配内存、不能执行阻塞操作、不能做耗时过长的处理、尽量避免print()。如果你在回调里尝试创建对象、拼接字符串,很可能看到memory allocation failed, allocating in ISR not allowed之类的错误,严重时直接死机。
我的解决方案就是把回调职责压缩到最小:只做时间比较、计数器累加、布尔标志置位。所有可能耗时的恢复动作全部延后到主循环执行。这样既保证了定时器回调的安全,也避免了恢复动作打断中断时序。这个设计是整个看门狗可靠性的基石,强烈建议照做,不要图省事直接在回调里处理恢复。
5.4 恢复动作执行时的互斥保护
还有一类坑出现在恢复动作本身执行期间。假如看门狗已经置位了recovery_pending,但主循环还没执行到process_recovery_events(),此时任务如果阴差阳错地又跑了一次并且喂了狗,会不会导致恢复动作被取消?
代码里我对这个场景做了保护:_check_all()遇到recovery_pending的任务直接跳过巡检,避免重复置位。但主循环的process_recovery_events()执行时,如果任务又喂狗了,它会在执行恢复动作前被清掉吗?不会,因为feed()只更新心跳时间戳和fault_count,不会动recovery_pending。也就是说,一旦本轮故障触发恢复,恢复动作无论如何都会执行完,保证不遗漏。
如果项目里有多个任务同时触发恢复,process_recovery_events()会顺序处理所有带recovery_pending的任务,不会互相干扰。但要注意,_execute_recovery()里的machine.reset()是终极大招,调用后系统立即复位,后面的任务恢复动作都不再执行,这是符合预期的——系统都重启了,其他任务自然重新初始化。
实际项目中还有一个建议:恢复动作里尽量不要依赖外部资源,尤其是网络、文件系统这类可能也处于异常状态的服务。比如reset()里如果尝试打日志到文件,而文件系统因为故障已经锁死,恢复逻辑就会被阻塞甚至抛异常。最稳妥的做法是恢复动作只做内存级的状态重置,不做 I/O,打印日志输出到串口就好。
结尾:一点实操体会
整套软件看门狗调下来,我最深的感受是:看门狗本身不难写,难点在于想清楚“什么算是故障”和“用什么方式恢复”。喂狗位置、超时阈值、恢复等级这些参数,必须结合业务场景反复打磨,没有一套参数能通吃所有项目。我建议你拿到这套代码后,先不要直接上真机,而是像我演示那样构造一个模拟挂起的任务,把故障注入点、喂狗点、恢复点全部跑通看日志,确认看门狗的行为符合预期了再嵌入到真实业务里。最后再分享一个实战小技巧:把看门狗注册、恢复日志统一加一个固定前缀,比如[WDT],线上问题排查的时候用串口工具过滤这个前缀,就能快速定位设备在哪个环节出现过任务级故障、走了哪一级恢复。这对提升运维效率帮助很大。