MicroPython轻量级任务调度器:让树莓派Pico从容处理多任务

发布时间:2026/9/7 12:49:30
MicroPython轻量级任务调度器:让树莓派Pico从容处理多任务 第一次拿到树莓派 Pico很多人第一反应是插上电脑跑一段 MicroPython点个 LED 就算入门了。结果等真要做点实际小项目——比如一边控制舵机、一边扫按键、再在 OLED 上刷两行状态——立刻就卡住了。MicroPython 不像 PC 上的 Python没法轻轻松松开十个线程跑着跑着就内存溢出、优先级乱套、代码糊成一团。这个问题的标准解法就是用定时器搭一个轻量级任务调度器让单核 MCU 也能从容地“同时”干好几件事。这篇文章我直接把我自己在 Pico 上用的调度器思路、代码和踩坑记录全写出来。不需要 RTOS不依赖复杂的异步框架就靠一个硬件定时器产生心跳再用主循环轮询任务表简单粗暴但非常稳定。适合刚上手 MicroPython、想从“点灯”进阶到真正做小项目的开发者参考。1. 为什么单核 MCU 也需要“多任务”1.1 MicroPython 的并发困境很多从 PC 端 Python 转过来的朋友下意识会用threading来写多任务。MicroPython 确实提供了_thread模块Pico 的双核 RP2040 也支持跨核启动线程但实际用下来坑非常多线程创建开销大、多个线程同时访问 I2C 或 UART 时会互相踩踏、内存碎片导致莫名重启、调试难度直线上升。我早期做过一个同时驱动舵机和读取 MPU6050 的项目用双线程写到后面最痛苦的已经不是功能逻辑而是“线程 A 往内存里写的值线程 B 什么时候才能读到”。真正的关键点在于单片机上的多数任务并不需要真并行它只需要“看起来像同时发生”。二十毫秒刷新一次舵机角度、二十毫秒扫描一次按键、两百毫秒刷新一次 OLED、五百毫秒翻转一次 LED这些任务交错执行对人眼来说就是完全并发的效果。而这正是任务调度器最擅长的场景。1.2 三种多任务方案的取舍我在 Pico 上实际尝试过三种做法简单对比一下方案优点缺点适合场景_thread多线程真并行、双核可用资源开销大、共享数据要加锁、调试痛苦数据采集与处理完全解耦的大型项目asyncio协程代码直观、不需要中断必须所有任务都写成协程、await习惯难改网络请求、传感器读取这类 IO 密集型任务定时器 主循环调度轻量、稳定、可预测性高任务代码必须快速返回、不适合超长耗时任务绝大多数传感器轮询、按键、显示刷新场景我最终长期使用的是第三种。原因也很简单它把“什么时间该跑什么任务”这个逻辑从业务代码里剥离出来了。以前写if now - last_time 100: refresh_screen()这种代码写在两三个任务里还能忍写到第六七个任务时整个主循环已经没法看了。用调度器之后主循环变得极其干净。1.3 调度器的核心思路心跳硬件定时器的本质就是每隔固定时间产生一次中断。把这个中断当成“心跳”每次心跳时在中断服务函数里把当前的全局计数器加一。主循环只需要判断每个任务设定的时间阈值有没有到到了就执行对应的函数。这样一个极其简单的机制就构成了一张精确的任务时间表。这个设计有一个很关键的好处时间基准完全由硬件定时器保证就算主循环里某个任务临时卡了 3 毫秒心跳计数依然在后台精准跳动不会累计误差。这是纯time.ticks_ms()做不到的——因为time.ticks_ms()需要主循环自己不断调用一旦主循环被某个操作卡住整个时间系统全部失灵。2. 调度器的整体设计与数据模型2.1 一个任务节点需要哪些信息要把调度器写得通用先想清楚“一个任务”由哪几部分组成。我最终抽象出这几个字段name任务名调试时打印用func要执行的函数对象period执行周期单位与 tick 一致我习惯用毫秒last上次执行时的心跳计数oneshot是否只执行一次enabled是否启用用于暂停/恢复我最开始没加name字段后来在项目里发现任务一多排查问题时根本分不清“是哪个函数在报错”每次都要数代码。加了名字之后日志输出立刻友好很多。2.2 为什么用“心跳计数差”而不是“当前时间戳”很多初版调度器会写成if time_now - last_time period: ...直接用毫秒时间戳。这在逻辑上没问题但有一个隐患如果任务周期非常短比如 1 毫秒而time.ticks_ms()的取模回绕周期只有约 25 天时间差计算在高频运行下偶尔会踩到边界条件。更重要的是每次循环里都调用time.ticks_ms()本身有函数调用开销虽然不大但高频检查时没必要。用硬件中断里的自增计数器就没有这个顾虑。中断每 1 毫秒触发一次计数器从 0 一直往上加只要用 Python 的整数类型根本不用考虑溢出。计算任务是否到期就是判断now - last period简单直接。2.3 为什么中断服务函数只做“最小工作”这是整个调度器最核心的设计原则中断服务函数ISR里绝对不执行任何用户任务。很多人第一次写定时调度时会直接在 Timer 回调里写func()图省事。但你在 ISR 里做的事情越多对主循环的打扰就越严重而且 MicroPython 在中断上下文中无法安全执行很多操作比如time.sleep、I2C 通讯、较大的字符串拼接、复杂的内存分配。我在调试时遇到过一种非常隐蔽的崩溃某个任务的函数内部调用了一个库函数恰好触发了 MicroPython 的垃圾回收而在中断上下文里 GC 会抛出MemoryError。那次排查花了我整整一个下午最后把任务迁到主循环执行问题立刻消失。所以我的调度器设计中ISR 永远只做两件事计数器加一、置一个标志位。真正干活全部在主循环里。3. 完整实现一个自带暂停、单次执行与优先级排序的调度器3.1 基础代码调度器核心类下面这段代码就是我目前在多个项目里稳定运行的任务调度器去掉了一些项目特定代码保留核心结构import machine from machine import Timer class Scheduler: def __init__(self, tick_ms1): self.tick_ms tick_ms # 心跳周期毫秒 self._ticks 0 # 心跳计数器 self._tick_flag False # 心跳标志位 self._tasks [] # 任务列表 self._timer Timer() # 硬件定时器 def _isr(self, timer): # 中断函数只做最轻量的事 self._ticks 1 self._tick_flag True def start(self): # 启动硬件定时器产生周期心跳 self._timer.init(modeTimer.PERIODIC, periodself.tick_ms, callbackself._isr) def stop(self): self._timer.deinit() def register(self, func, period, nametask, oneshotFalse, enabledTrue, priority0): # 注册一个任务返回任务句柄方便后续暂停/恢复 task { name: name, func: func, period: period, last: 0, oneshot: oneshot, enabled: enabled, priority: priority, } self._tasks.append(task) # 按优先级从高到低排序数值越小优先级越高 self._tasks.sort(keylambda t: t[priority]) return task def pause(self, task): task[enabled] False def resume(self, task): task[enabled] True task[last] self._ticks # 重新从当前时刻开始计时 def loop(self): # 主循环中反复调用调度器的心脏 if not self._tick_flag: return self._tick_flag False now self._ticks for task in self._tasks: if not task[enabled]: continue if now - task[last] task[period]: task[last] now try: task[func]() except Exception as e: print(Task [%s] error: %s % (task[name], e)) if task[oneshot]: task[enabled] False这段代码的核心逻辑可以用几句话说明start()初始化硬件定时器每隔tick_ms毫秒触发一次_isr()在中断里只是简单地把_ticks加一。loop()检查到心跳标志位后遍历所有任务判断当前计数减去上次执行计数是否达到周期如果达到就执行任务函数。单个任务执行出错不会影响其他任务只会打印一条错误信息。3.2 使用示例三个任务并行跑我写一个最直观的示例同时跑三个任务一个控制 LED 以 500ms 周期闪烁一个在 1000ms 周期打印一段“串口心跳”一个只执行一次的单次任务在 3 秒后打印提示。from machine import Pin import time led Pin(25, Pin.OUT) sched Scheduler(tick_ms1) def blink_led(): led.toggle() def print_heartbeat(): print(heartbeat:, sched._ticks) def one_shot_task(): print(one shot task executed!) # 注册任务 sched.register(blink_led, period500, nameblink) sched.register(print_heartbeat, period1000, nameheartbeat) single_task sched.register(one_shot_task, period3000, oneshotTrue, namesingle) # 启动调度器 sched.start() # 3 秒后又想取消单次任务可以直接 pause # 实际运行中把下面的 time.sleep 换成 sched.loop() 的死循环 while True: sched.loop_time time.ticks_ms() sched.loop()注意这里while True里只调用了sched.loop()所有任务逻辑都已经封装在外部函数里。主循环异常干净新加一个任务只需要新增一个函数加一行注册代码不需要再动主循环的任何逻辑。这就是调度器带来的最大工程价值。3.3 关于优先级排序的说明register里的priority参数默认都是 0。需要调整时数值越小越先执行。考虑到 MicroPython 的字典和列表操作在任务数量少时开销很低任务数量在几十个以内直接遍历完全没问题。真正的微秒级实时任务比如 PWM 波形的控制不应该放在这个调度器里而应该直接用硬件外设去完成这也是我一直强调的思路调度器解决的是“吞吐型”任务实时性要求极高的任务交给硬件模块。3.4 中断频率选择为什么默认 1ms我在不同项目里试过 100 微秒到 10 毫秒之间的多种心跳频率。最终稳定使用的默认值是 1 毫秒。原因有三点第一1ms 足够覆盖绝大多数传感器的轮询需求按键去抖、舵机角度更新、屏幕刷新这些场景都在几十毫秒到几百毫秒之间。第二RP2040 的 MicroPythonTimer在 1ms 下实测抖动非常小稳定可靠。第三如果心跳设置到 0.1ms中断过于频繁主循环的执行会被打断得七零八落反而导致任务执行时间产生明显抖动。当然如果你只是做按键扫描和 LED 闪烁tick_ms5也完全够用还能进一步降低 CPU 开销。具体取值要在项目里实测结合你自己的主循环耗时来定。4. 真实场景联动舵机 按键 OLED 的组合调度单独展示调度器看不出威力真正让我彻底放弃手写时间判断的是一个同时包含舵机控制、按键去抖、OLED 状态刷新的项目。这也是很多做小车、机械臂、智能家居小面板的朋友都会遇到的典型场景。4.1 舵机控制PWM 交给硬件调度器管时机舵机控制这个热搜词非常经典。SG90 舵机需要一个 50Hz、占空比 0.5ms~2.5ms 的 PWM 信号。在 Pico 上用 MicroPython 的machine.PWM就可以直接生成根本不需要 CPU 持续参与。我们需要调度的只是“什么时间把舵机转到什么角度”这个决策逻辑。from machine import Pin, PWM # 舵机初始化PWM 频率 50Hz servo PWM(Pin(15)) servo.freq(50) def set_servo_angle(angle): # 角度 0~180 映射到占空比 0.5ms~2.5ms # 50Hz 下 16bit 满占空比对应 6553520ms 周期对应 1ms 3276.75 duty 1638 int(angle / 180 * 6553) duty max(1638, min(8191, duty)) servo.duty_u16(duty) def servo_slow_move(): # 以一个固定步进角度平滑转动舵机 global current_angle, target_angle if current_angle target_angle: current_angle 1 set_servo_angle(current_angle)这里servo_slow_move被注册成 20ms 周期任务后舵机每 20ms 变化 1 度从 0 度转到 180 度大概需要 4 秒钟非常平滑。如果你手写主循环这个效果当然也能实现但一旦同时再处理别的事情循环一卡舵机角度更新就变得一顿一顿。放在调度器里只要主循环能在 20ms 内转一轮舵机就能保持流畅。4.2 按键扫描中断只记录调度器做去抖按键处理是单片机项目里最常遇到的需求之一。如果用外部中断去做完整处理很容易在中断里出现多次触发。我的做法是按键的Pin.IRQ中断只记录一个事件标志和时间戳然后由调度器里的一个高频任务去统一处理。from machine import Pin import time btn Pin(16, Pin.IN, Pin.PULL_UP) btn_pressed False btn_debounce_last 0 def btn_isr(pin): global btn_pressed # 中断里只记录一次 btn_pressed True btn.irq(triggerPin.IRQ_FALLING, handlerbtn_isr) def scan_button(): global btn_pressed, btn_debounce_last if not btn_pressed: return # 去抖距离上次触发至少 20ms now time.ticks_ms() if time.ticks_diff(now, btn_debounce_last) 20: return btn_debounce_last now btn_pressed False print(Button pressed, start servo move!) target_angle 180 if target_angle 0 else 0这里的精髓是中断函数btn_isr只做最简单的翻转标志位操作耗时微乎其微。真正需要时间判断去抖、需要调用打印、需要触发舵机转动的复杂逻辑全部放到了scan_button任务里。我把scan_button注册为 5ms 周期任务执行频率足够高人按按键时不会丢失事件。4.3 OLED 刷新绝不在中断里操作 I2C很多用 ESP32 S3、树莓派 Pico 玩 MicroPython 的朋友都会配一块 SSD1306 OLED 小屏因为热点词里就有“esp32 s3 oled micropython”。这个屏幕走 I2CMicroPython 有现成的ssd1306驱动库使用非常方便。但它有一个大坑I2C 通讯耗时较长绝对不能在定时器中断上下文里直接刷新屏幕否则轻则屏幕闪烁、任务阻塞重则直接死机。正确姿势是把显示刷新也作为一个低优先级任务以 200ms 周期执行。from ssd1306 import SSD1306_I2C import machine i2c machine.I2C(0, sclmachine.Pin(9), sdamachine.Pin(8), freq400000) oled SSD1306_I2C(128, 64, i2c) def refresh_oled(): oled.fill(0) oled.text(Servo: %d deg % current_angle, 0, 0) oled.text(Target: %d % target_angle, 0, 16) oled.text(Tick: %d % (sched._ticks), 0, 32) oled.show()refresh_oled注册为 200ms 周期任务200ms 足够主循环其他任务正常完成同时人眼看起来很流畅。4.4 完整任务注册sched.register(servo_slow_move, period20, nameservo_move) sched.register(scan_button, period5, namebtn_scan) sched.register(refresh_oled, period200, nameoled_refresh, priority10) sched.register(blink_led, period500, nameled_blink) sched.start() while True: sched.loop()这段代码启动后整个系统就是一个小型“状态机 时间驱动”的组合体。舵机平稳转动、按键实时响应、屏幕定期刷新、LED 按自己的节奏闪烁相互之间互不干扰。要加新功能只需要写一个新任务函数注册到调度器里不需要动主循环。5. 踩坑记录与排查清单5.1 高频踩坑一在任务里使用 time.sleep我见过很多新手在任务函数里写time.sleep(0.05)想让这个任务“歇一会儿”。这是大忌。任务调度器里的所有任务都运行在主循环中任何一个任务阻塞其他所有任务全部停滞。我在实际项目里排查过一个问题按键按下后 LED 要延迟 2 秒才闪烁原因是某个任务里的time.sleep(1)把整条主循环拖住了。解决办法是把延迟状态化。如果需要“延迟 2 秒后做什么”记录目标时间戳在任务里判断now target再执行而不是去 sleep。如果确实有需要等待外部设备响应的长操作比如网络请求要么把这个操作改成非阻塞模式要么把它单独放到_thread里不要放进调度器。5.2 高频踩坑二共享变量的可见性调度器里的任务函数全部在同一个线程中执行这反而是它的好处——不需要考虑线程安全问题一个普通变量就能在多个任务之间共享。但有一个问题需要注意MicroPython 的局部变量访问比全局变量快很多你在循环里反复读取一个全局变量时建议先赋值给局部变量任务结束再写回全局。我在servo_slow_move里就用了这个方法避免高频情况下全局访问的开销。如果任务和中断服务函数共享变量比如btn_pressed只要中断里只做“赋值”或“翻转”主循环里只做“读取”一般是安全的。但如果要传一个更复杂的数据结构建议用bytearray之类的不易被垃圾回收影响的容器或者直接用列表里固定位置存储。5.3 高频踩坑三Timer 周期单位在各平台不一致MicroPython 的Timer在不同开发板上的参数存在差异。树莓派 Pico 的machine.Timer的period单位是毫秒而 ESP32 上使用Timer时需要指定timer id初始化方式也不一样比如Timer(0)或Timer(1)。我在 ESP32 S3 上移植这个调度器时只改了一行 Timer 初始化的代码其他完全通用。这也说明把调度器封装成独立类是值得的跨平台迁移时不需要改任务逻辑。5.4 高频踩坑四任务执行时间超过任务周期当任务的执行时间超过了它自己的周期或者多个任务的执行时间总和超过了心跳周期调度就会出现“积压”现象。表现为任务执行频率低于预期打印日志发现有些周期任务跳到高峰后又突然停滞。排查方法是每次任务开始和结束时打点用time.ticks_diff()记下耗时看是哪个任务吃掉了大部分时间。我的经验是任何任务的单次执行时间控制在一毫秒以内是最理想的超过这个量级就该考虑优化算法或者把任务拆小。5.5 问题速查表现象可能原因解决思路任务整体变慢、间隔不稳定某个任务里用了time.sleep或长耗时操作改成状态机 时间戳判断打印错误cant sleep in ISR在定时器回调里直接调用了sleep回调只做计数和置标志任务挪到主循环单次任务执行后仍在触发oneshot没设置或不生效检查注册参数执行后enabledFalse舵机角度抖动主循环被其他任务阻塞更新不及时增大舵机任务周期或降低其它任务耗时OLED 屏幕花屏或卡死在中断上下文操作 I2C刷新任务放到主循环调度器按键触发一次却执行多次去抖时间不够长增加去抖时间到 30ms~50ms写在最后任务调度器这个东西本质上就是把“时间管理”从业务逻辑里抽离出来让每一个功能模块只关心自己该干什么不用关心别人什么时候干。我刚接触嵌入式时总想着用更“高级”的方式解决并发问题试过线程、试过异步库最后发现最可靠的反而是这个用硬件定时器当心跳、用主循环做轮询的土办法。它不炫酷但它稳定、可预期、好调试。后来我在很多项目里都复用了这套代码换板子、换传感器、加功能主循环始终就是那一行sched.loop()。如果你也被 MicroPython 的多任务问题折腾过不妨先抄这个方案试一试。