树莓派Pico轻量级多任务调度:基于定时器的MicroPython时间片轮转实现

发布时间:2026/9/6 8:51:33
树莓派Pico轻量级多任务调度:基于定时器的MicroPython时间片轮转实现 把树莓派 Pico 接上一个舵机、一个 OLED、一个温湿度传感器之后我最头疼的一件事就是代码里的time.sleep()越来越多LED 闪烁频率开始忽快忽慢按键响应也变成“爱答不理”。问题不在硬件而是我把一堆需要周期性执行的任务全部串行写进了主循环。接触过 51 定时器、STM32 定时器中断的朋友应该能秒懂这种感觉——裸机程序一旦有多个周期任务就必须引入一个“任务调度器”。这篇文章我用 MicroPython 在 Pico 上实现了一个极轻量的基于定时器时间的多任务调度骨架不依赖_thread不引入 RTOS能跑时间片轮转、周期任务、带中断标志位的准抢占式扩展。写给想把多个周期任务干净地跑在一个 Pico 上的朋友也写给刚从 Arduino/STM32 转过来、想了解 MicroPython 多任务写法的开发者。读完你可以直接抄走整套代码也能理解它背后的定时器原理。1. 单核单片机上的“多任务”到底是怎么一回事1.1 为什么裸机程序总在“因为一段延时全盘卡住”如果你以前写过 51 或者 STM32 的裸机程序大概率经历过这种场景主循环里既要刷新数码管又要扫描按键还要处理串口数据。一开始为了简单直接在代码里加delay_ms(10)做按键消抖加delay_ms(200)做显示刷新加delay_ms(1000)做传感器采集。结果就是程序表面在“多任务”实际所有任务都被绑在一根绳子上一秒一秒地排队你按一下按钮可能 500ms 之后才有反应。到了树莓派 Pico 的 MicroPython 上很多人还是沿用这套思路。尤其是从 Arduino 转过来的朋友习惯了loop()里挨个调函数很容易写出这种代码while True: led.toggle() time.sleep(0.5) read_temp() time.sleep(0.2) refresh_oled() time.sleep(0.1)这段代码的致命问题在于time.sleep()会把 CPU 整个占用期间按键按下去了没人管串口数据来了没人收PWM 输出虽然硬件还在跑但你的业务逻辑完全处于“失聪”状态。我做过一个很小的实验让 LED 每 500ms 翻转一次同时让 OLED 每 200ms 刷新一次中间穿插 DHT11 温湿度读取。DHT11 这个传感器读取一次要阻塞几十毫秒甚至更久结果 LED 的闪烁周期肉眼可见地变得毫无规律。真正跑起来以后我才意识到在单片机这种单核裸机环境里所谓“多任务”本质就是把这个 CPU 的执行时间切分成很多小片轮流分给不同的事情干。1.2 时间片、抢占与协作三个底层概念一次说清要理解任务调度器先要分清两个最容易被混淆的概念协作式调度和抢占式调度。协作式调度的核心是“每个任务自觉按时返回”。任务之间靠一个调度循环手动切换当前任务如果不主动让出 CPU其他任务就一直等。好处是逻辑简单、资源开销小坏处是一个任务写了个死循环或者长时间的阻塞延时整个系统就崩了。抢占式调度的核心是“用定时器中断强制切换”。不管当前任务跑到哪一行只要定时器中断一进来CPU 就立刻跳去执行下一个任务。这就是 RTOS 和普通裸机最大的区别。它保证高优先级任务能被准时执行但代价是上下文切换和内存开销都更大移植和调试复杂度也更高。在 MicroPython 解释器环境下我们其实很难做成真正的抢占式线程切换因为解释器本身有一套变量和对象模型强行从中断里切换任务很容易把 Python 的动态对象搞乱。所以在实际工程里大家更常用的是一条中间路线定时器中断负责“准时提醒”标志位置位主循环负责“执行任务”。这样既有了定时器的可靠性又避免了在中断里操作 Python 对象的风险。我后面要讲的进阶调度器走的就是这条路。1.3 什么场景值得手写调度器什么场景直接上 RTOS很多人一听说“任务调度”第一反应是那不如直接移植 FreeRTOS。这话对一半。Pico 本身确实能跑 FreeRTOS但如果你用的是 MicroPython再塞一个 RTOS 的思维进去反而别扭MicroPython 解释器已经把硬件都封装好了底层任务切换的意义不大你真正需要解决的只是“多个周期任务如何错开时间”的问题。我的判断标准是任务数量 3~5 个周期性强不需要严格实时性手写调度器完全够用任务数量超过 10 个或者任务之间存在复杂的阻塞/唤醒关系考虑 RTOS 或者更成熟的调度框架需要硬实时响应比如电机电流环MicroPython 本身就不合适应该换 C 语言加硬件定时器中断去写。手写调度器的最大价值不是“造轮子”而是让你彻底理解时间片、定时器中断、非阻塞编程这几个核心概念。以后不管是看 RTOS 源码还是用现成的调度框架你都会心里有底。2. 定时器资源摸底Pico 上谁在提供“时间心跳”2.1 RP2040 的系统定时器与 MicroPython 的 time.ticks_ms()树莓派 Pico 的芯片叫 RP2040它内部有一个 64 位的系统定时器上电后就以微秒为单位一直累加计数直到断电才清零。这个计数器是整个系统的时间基准MicroPython 的time.ticks_ms()、time.ticks_us()底层都是从这个计数器派生出来的。用time.ticks_ms()获取当前毫秒数看起来很简单但它有一个很容易踩坑的特性这个返回值不是无限增长的它会周期性回绕。MicroPython 文档里明确说明ticks_ms()的返回值范围是 0 到 2^30-1约等于 12.4 天之后就会归零重来。如果你在代码里直接写if now - last_run period一旦回绕发生这个判断就会出错任务突然停摆或者疯狂触发。正确做法是永远用time.ticks_diff()和time.ticks_add()来做时间比较和加减MicroPython 内部会正确处理回绕。这也是我下面所有调度代码的基础。2.2 machine.Timer 中断与 ticks_ms 轮询的取舍在 Pico 上做周期任务有两套时间方案可选。方案 A纯 ticks_ms 轮询。主循环每次运行都检查当前时间看看哪些任务到期了。实现最简单不依赖任何中断但时间精度取决于主循环的执行频率。主循环跑得快精度就高主循环里有一个任务耗时 50ms那其他任务的 10ms 精度就完全泡汤。方案 Bmachine.Timer 中断。用硬件定时器产生一个周期中断比如每 1ms 中断一次在中断回调里更新任务状态。由于中断可以打断主循环所以无论主循环当前在干什么任务到期的“打点时刻”是准的。但中断回调里只能做简单的标志位操作真正耗时的处理必须丢回主循环。Pico 的machine.Timer在官方固件里可以直接创建定时器实例有的固件版本需要显式指定编号有的直接Timer()无参创建也可以。我在实际使用中优先选择中断方案来做调度打点因为几行代码就能把“调度时刻”和“任务执行”解耦后面扩展优先级、累计丢失次数都方便。2.3 为什么不推荐直接用 _thread 做多线程有朋友看到 Pico 是双核 RP2040第一反应是MicroPython 支持_thread我直接开两个线程不就行了我一开始也这么干过实测体验并不好。MicroPython 的线程模型和 CPython 一样受 GIL 限制两个 Python 线程不能真正并行执行字节码只能通过时间片轮转交替跑。而且线程共享变量时稍有不慎就会出现数据错乱需要一个Lock来保护。在 Pico 这种内存只有 264KB 的平台上_thread还需要给每个线程分配独立栈空间资源吃紧。更关键的是_thread的调度完全黑盒化你没法控制哪个任务在哪个时刻跑也就无法精确保证某个周期任务的节拍。而手写调度器时间片、优先级、首次延迟全部由你自己掌控。对于大部分 Pico 项目基于定时器的协作式调度比线程模型靠谱得多。3. 核心实现写一个最轻量的时间片轮转调度器3.1 任务对象的本质只需要记住“下次该跑的时间”如果你只是想在 Pico 上同时跑几个周期任务其实不需要任何复杂数据结构。每个任务只需要保存三样东西自己的执行周期period_ms自己要执行的那个函数handler下一个执行时刻next_run。微控制器毕竟不是 PC我们不需要保存上下文、不需要切换栈、不需要维护就绪队列。因为这是协作式调度任务之间轮流靠“时间”自然错开谁到点了谁就执行。我封装了一个Task类import time class Task: def __init__(self, period_ms, handler, delay_ms0, nametask): self.period_ms period_ms self.handler handler self.name name self.next_run time.ticks_add(time.ticks_ms(), delay_ms) def is_due(self, now): return time.ticks_diff(now, self.next_run) 0 def run(self, now): self.handler() self.next_run time.ticks_add(now, self.period_ms)关键点有两个。第一next_run的初始值可以用delay_ms控制这样我可以在启动时让不同任务错开第一个执行点避免所有任务在同一个时间片爆发减少峰值卡顿。第二run()里计算下一次执行时间用的是now也就是“从本次实际执行开始再等一个周期”。如果某个 handler 执行超时了任务的下一次触发也会顺延不会出现“上一轮还没跑完下一轮又到点”的堆积问题。另一种常见写法是next_run ticks_add(self.next_run, self.period_ms)让任务严格按固定节拍补跑。实测下来在 MicroPython 这种非实时环境下顺延式比补跑式稳定得多至少不会出现一个慢任务卡住其他任务后突然连跳几次的诡异现象。3.2 调度循环几行代码就能实现的“任务分发”有了Task类调度器本身简单得让你怀疑是不是漏了什么。class Scheduler: def __init__(self): self._tasks [] def add(self, task): self._tasks.append(task) def schedule(self, now): for task in self._tasks: if task.is_due(now): task.run(now)主循环就一行sched Scheduler() sched.add(Task(500, blink, nameblink)) sched.add(Task(1000, report, namereport)) sched.add(Task(20, scan_key, namekey)) while True: sched.schedule(time.ticks_ms())每次循环都遍历任务列表检查是否到期到期的就执行。由于单个 task 的 handler 通常只有几行代码整个循环跑一圈就是几十微秒配合ticks_ms()的毫秒粒度完全够用。要注意的是这个调度器没有“睡眠”概念它永远在空转跑循环。别担心功耗Pico 裸板跑 Python 功耗本来就是几十到一百毫安级别如果真要做低功耗再配合lightsleep()去处理那是另一篇文章的话题。3.3 完整示例LED 闪烁 周期日志 非阻塞按键扫描为了让你直观感受到这套调度器的效果我写了一个完整的示例。开发板上自带一颗 LEDPin 25把按键接在 GP0 和 GND 之间上拉输入。三个任务LED 每 500ms 翻转一次系统日志每 2000ms 打印一次按键扫描每 5ms 执行一次并做 20ms 消抖。from machine import Pin import time led Pin(25, Pin.OUT) key Pin(0, Pin.IN, Pin.PULL_UP) def blink(): led.toggle() def report(): print(system alive, ticks:, time.ticks_ms()) # 非阻塞按键扫描配合调度器轮询 _prev_key 1 _prev_debounce_ticks time.ticks_ms() def scan_key(): global _prev_key, _prev_debounce_ticks state key.value() now time.ticks_ms() if state ! _prev_key: if time.ticks_diff(now, _prev_debounce_ticks) 20: if state 0: print(key pressed) _prev_key state _prev_debounce_ticks now sched Scheduler() sched.add(Task(500, blink, nameblink)) sched.add(Task(2000, report, namereport)) sched.add(Task(5, scan_key, namekey)) while True: sched.schedule(time.ticks_ms())按键消抖这一段很典型。以前写阻塞式消抖通常要time.sleep_ms(20)这 20ms 就把其他任务全部卡住了。现在消抖逻辑改成状态机式每次只检测一次电平变化然后记录时间到 20ms 后再确认一次整个过程没有任何sleep完美融入调度器。4. 进阶方案定时器中断驱动 优先级 一次性任务4.1 MicroPython 定时器回调的“禁区”上面这版调度器已经够用了但它有个先天弱点如果某个任务的 handler 执行时间过长下一轮schedule()被推后依赖时间精度的任务就会出现肉眼可见的抖动。所以我想把“打点”这件事从主循环里彻底解耦出去交给定时器中断。Pico 支持machine.Timer创建周期定时器最小周期可以到 1ms。中断回调在 MicroPython 里有严格的限制这一点必须反复强调回调处理器里绝对不能调用print()因为 print 会申请内存、可能加锁轻则延迟重则直接触底复位回调里绝对不能进行变量赋值以外的大操作比如构造字符串、创建列表、操作 I2C回调里绝不能time.sleep()回调里抛异常整个 MicroPython 环境可能直接崩溃报出一堆类似Fatal error during exception handling的信息。很多第一次用 machine.Timer 的朋友习惯把业务逻辑整个塞进回调结果就是板子无缘无故重启。这个问题不是 Pico 独有的ESP32、STM32 的 MicroPython 也一样本质是 Python 解释器在中断上下文里做内存管理不安全。4.2 中断里只做检查主循环只做执行安全的中断驱动做法是在中断回调只更新任务的到期标志位真正的 handler 留在主循环里执行。from machine import Timer import time class SoftTimerTask: def __init__(self, period_ms, handler, delay_ms0, nametask): self.period_ms period_ms self.handler handler self.name name self.next_run time.ticks_add(time.ticks_ms(), delay_ms) self.due False self.missed 0 def check(self, now): if time.ticks_diff(now, self.next_run) 0: self.next_run time.ticks_add(now, self.period_ms) self.due True self.missed 1 def run(self): if self.due: self.due False self.missed 0 self.handler()调度器的主循环里只有一行tasks [SoftTimerTask(10, health_check, namehealth), SoftTimerTask(200, refresh_oled, nameoled)] def tick(t): now time.ticks_ms() for task in tasks: task.check(now) timer Timer() timer.init(period1, modeTimer.PERIODIC, callbacktick) while True: for task in tasks: task.run()我把定时器周期设为 1ms这样任务的到期检查精度是 1ms。中断回调里只有整数比较、加法和一个布尔量置位这些操作在 MicroPython 中断环境下是安全且快速的。主循环里各取所需谁due为 True 就执行谁。这个结构带来一个额外好处我可以知道任务是否“错过多次”。如果主循环因为某个耗时任务卡了 30msmissed会累加到 3 甚至更高。在低功耗或者性能紧张的设备上这组计数可以直接作为系统健康指标来监控。4.3 优先级与一次性任务扩展如果需要让某些任务优先执行最简单的做法是给SoftTimerTask加一个优先级字段然后每次主循环按优先级排序后再执行class SoftTimerTask: def __init__(self, period_ms, handler, delay_ms0, priority0, nametask): self.priority priority ... # 执行前按优先级从小到大排序数字越小优先级越高 tasks.sort(keylambda t: t.priority) while True: for task in tasks: task.run()注意枚举的顺序决定了同一次“到点检查”之后哪个任务先跑。比如 10ms 的健康检查和 200ms 的 OLED 刷新同时到点如果传感器采样任务优先级更高它会先执行OLED 刷新可能因此延后几百微秒但这通常不影响业务。一次性任务也很简单在run()里执行完 handler 之后把任务从列表里移除或者给一个oneshot标志def run(self): if self.due: self.due False self.handler() if self.oneshot: self.dead True这个设计很适合做“延时启动”动作比如开机 5 秒后自动校准一次舵机或者设备运行 1 小时后把某个标志位置位。以前用sleep()实现这种功能会把整个循环卡住现在它只是调度器里的一个普通任务。5. 踩坑实录调度器跑起来之后的问题与排查链路5.1 定时器回调里 print板子直接重启我第一次把machine.Timer的周期设为 1ms然后在回调里写了一句print(.)想看看定时器是不是在跑。结果板子直接重启连KeyboardInterrupt都拦不住反复试了几次都一样。排查过程是这样先怀疑是 Timer 外设初始化问题把回调注释掉程序正常恢复回调但去掉 print程序也正常。这说明问题出在 print 本身。MicroPython 的 print 会调用解释器的字符串格式化中间有内存分配内存分配在中断上下文里是不可重入的。而且 print 输出到 USB CDC 串口还要等待 USB 端点缓冲一旦 USB 正在忙这个回调就可能卡死。这里我总结出一条硬性规则中断回调里只许做“写标志位、简单计数、短小整数运算”这三件事其他一律移到主循环。如果实在需要调试用一个环形缓冲在中断里塞整数状态主循环再统一打印。5.2 ticks_ms 回绕导致的任务“突然停摆”还有一个坑藏得很深。刚开始我写的判断是if now - task.next_run task.period_ms一开始运行非常稳定但连续跑了大概 12 天之后如果你经常让板子挂在墙上一定会遇到任务突然全部“停摆”了。根因就是前面提到的ticks_ms()回绕。当系统计数从 2^30-1 回到 0 时now - task.next_run的数学结果会变成一个巨大的负值条件永远不成立调度器就“死”了。排查的方式是看时间戳当时打印出来的 ticks 值从 1073741800 多突然跳到 0 附近。修复方法很简单就是用ticks_diff(now, task.next_run) 0加减时间用ticks_add绝对不要自己用和-。一旦用了ticks_diff回绕会被 MicroPython 内部正确处理程序可以连续跑几个月不重启。这件事也提醒我MicroPython 这些 ticks 系列函数从来不是为了规范而规范的。5.3 一个任务阻塞所有任务排队延迟就算调度器写得再干净主循环里只要有一个 handler 里写了time.sleep(1)其他所有任务立刻被拖住表现出来就是按键延迟、LED 乱闪。这个问题的背后是“协作式调度”的天然约束任务主动让出 CPU。解决思路有几个把time.sleep()改成非阻塞等待。比如等待某个事件可以改成检查标志位和超时时间不到期就立刻返回。把耗时操作拆成状态机。比如 OLED 刷新时一次show()可能耗时几十毫秒如果要求不高可以降低刷新频率到 200ms每帧画面分两次绘制。如果一个任务确实无法避免长阻塞比如要串口发 1KB 数据可以考虑用 DMA 或后台硬件处理而不是 CPU 傻等。我在调试时经常用一个小技巧给每个任务包一层“统计执行时间”的装饰器在 handler 前后各取一次ticks_us()一旦发现某个任务耗时超过它的周期的 50%就立刻优化它。这个过程比想象的更能暴露问题很多任务表面看似简单实际一测耗时还挺惊人。5.4 调度抖动与 GC 的关联最后说一个不容易定位的抖动来源MicroPython 的垃圾回收GC。堆内存满了之后解释器会自动触发一次 GC这个过程可能持续几毫秒到十几毫秒。如果 GC 恰好发生在某个关键任务的执行节点附近调度器会观察到一次明显的延迟尖峰。我的排查方式是周期打印各任务的耗时发现每隔一段时间就出现一次 5~8ms 的异常峰值而且和业务代码无关。后来定位到是 GC。解决方案有几个层面尽量在程序启动时一次性把大的对象创建好避免运行中频繁分配和释放如果用的是官方固件可以微调 MicroPython 堆大小和 GC 阈值对实时性要求最高的任务比如舵机脉冲更新最好用 machine.PWM 等硬件外设直接输出不依赖 Python 任务的实时性。这一条经验告诉我MicroPython 不是实时系统任何“精确时间调度”都需要对解释器本身的开销有预期。调度器能做到毫秒级稳定节拍已经很好想达到微秒级应该考虑 C 语言或者直接操作硬件外设。6. 实战舵机、OLED、传感器三个周期任务同时跑6.1 场景拆分与任务周期分配前面讲了这么多原理和坑最后我拿一个真实项目把它们串起来。需求是这样的一个 SG90 舵机需要按平滑轨迹从 0 度转到 180 度再翻回来周期约 4 秒一个 128x64 OLED 屏幕显示当前温度和舵机角度需要每秒刷新 5 次左右一个 I2C 温湿度传感器比如 AHT20每 2 秒采样一次。三个任务如果串行写OLED 刷新一次 20ms、传感器读取一次 5ms加上各种打印整个系统跑下来会非常卡。我的分配方案是舵机目标角度计算每 50ms 一次。注意舵机本身是 PWM 硬件控制的一旦设置好占空比RP2040 的 PWM 外设会持续输出不需要 CPU 干预所以这个任务只是每隔 50ms 算一个新角度写入 PWM 寄存器。传感器采集每 2000ms 一次。AHT20 的 I2C 读取很快但为了和数据展示节奏匹配2 秒足够。OLED 刷新每 200ms 一次。屏幕只在传感器数据更新后才需要重绘200ms 足够平滑。我把这三个任务的delay_ms分别设为 0、50、100让它们的执行点错开避免三个任务在同一毫秒内全部到点。6.2 完整集成代码与细节说明下面是一个可以放到main.py里直接跑的版本传感器用 AHT20 的简化驱动逻辑重点看任务分配和调度器集成from machine import Pin, PWM, I2C import time # ---------- 硬件初始化 ---------- led Pin(25, Pin.OUT) servo PWM(Pin(15)) servo.freq(50) i2c I2C(0, sclPin(1), sdaPin(0), freq400000) # oled 需要把 ssd1306.py 放到 Pico 文件系统或者用车载驱动 # 这里省略 import重点看调度结构 def set_angle(angle): # 0~180 度映射到 0.5ms~2.5ms 脉宽PWM 频率 50Hz周期 20000us pulse_us 500 int(angle / 180 * 2000) duty int(pulse_us * 65535 / 20000) servo.duty_u16(duty) def read_temperature(): # 简化实际需要按 AHT20 数据手册读取 # 这里演示返回一个模拟值实际项目替换成真实读取 return 26.5 # ---------- 任务函数 ---------- angle 0 step 2 def update_servo(): global angle, step angle step if angle 180 or angle 0: step -step set_angle(angle) defrefresh_sensor(): # 温度放入全局变量供 OLED 任务读取 global temp temp read_temperature() def refresh_screen(): # 非阻塞绘制 刷新 # 实际使用时把这段替换成 ssd1306 的 fill/text/show pass # ---------- 注册调度器 ---------- sched.add(Task(50, update_servo, delay_ms0, nameservo)) sched.add(Task(2000, refresh_sensor, delay_ms50, namesensor)) sched.add(Task(200, refresh_screen, delay_ms100, nameoled)) while True: sched.schedule(time.ticks_ms()) led.toggle() # 用 LED 指示主循环还在跑这里最容易被忽略的是set_angle里的脉宽计算。SG90 舵机要求 50Hz、0.5ms~2.5ms 脉宽对应 0~180 度。RP2040 的 PWM 时钟经过分频后周期 20ms 对应duty_u16的范围是 0~65535所以 0.5ms 对应的 duty 约为 16382.5ms 约为 8192。如果你直接填一个 0~100 的百分比舵机大概率不会转到位。我当时在这上面耗了不少时间最后用示波器量 A 和 B 两个通道的波形才定位到问题。如果你的项目里有 OLED需要注意ssd1306模块默认不是我在使用官方固件而是需要把驱动文件放到 Pico 里。我建议选 128x64 的小尺寸刷新频率控制在 200ms 以上否则 I2C 通信本身就会占掉大量时间反而拖累调度器。6.3 把调度器扩展成可配置任务表的思路实际项目里任务会越来越多手动一行行sched.add还能接受但一旦超过 6~7 个我习惯把任务表定义成可配置的结构方便统一开关和调整周期TASK_TABLE [ {name: servo, period_ms: 50, handler: update_servo, delay_ms: 0}, {name: sensor, period_ms: 2000, handler: refresh_sensor, delay_ms: 50}, {name: oled, period_ms: 200, handler: refresh_screen, delay_ms: 100}, ] for item in TASK_TABLE: sched.add(Task(item[period_ms], item[handler], delay_msitem[delay_ms], nameitem[name]))如果你愿意还可以支持每个任务的“启用/禁用”标志、运行计数、错误回调。但我不建议一开始就把调度器设计得太花哨。MicroPython 项目最重要的是“跑得稳、看得懂”先把基础的时间片调度跑顺再按实际需求去增加功能。我这几套代码已经在好几个项目里跑过包括多路舵机控制、OLED 翻页显示、传感器循环采集、按键菜单交互稳定性都足够。唯一需要你记住的一条铁律是任何 handler 都不能长时间阻塞所有耗时操作要么拆开、要么降低频率、要么交给硬件外设。守住这条调度器基本不会坑你。