资讯动态

树莓派Pico多任务实践:定时器驱动的轻量级任务调度器

发布时间:2026/9/9 2:49:53 来源:尧图企业网站定制
1. 项目概述与核心思路先说结论树莓派 Pico 这块板子虽然只是个微控制器但在 MicroPython 环境下做多任务并没有想象中那么麻烦。这篇文章要聊的就是一个非常朴素但又极其有用的实现方式——用定时器驱动一个轻量级任务调度器让多个任务“同时”跑起来。MicroPython 默认是单线程执行的一个while True循环里如果塞了多个耗时操作比如读传感器、控制舵机、刷新 OLED就会遇到一个问题前面的代码卡住了后面的代码就没机会跑。有人会说“那用_thread多线程啊”但 Pico 只有两个核心MicroPython 的线程支持又受 GIL 限制再加上线程切换带来的内存开销在很多场景下反而更容易踩坑。定时器驱动的任务调度器解决的是另一类问题按固定的时间片去轮转执行任务每个任务只在自己的时间片里运行运行完就交出 CPU。说白了是把“时间”切成均匀的格子每个格子安排一个任务定时器负责叫醒下一个任务。这种方式在嵌入式场景里非常实用尤其是 IO 密集型、对实时性要求不极端苛刻的项目。这个方案适合谁适合已经在用 MicroPython 写 Pico、手上有好几个外设要一起管、又不想引入复杂 RTOS 的开发者。读完这篇文章你能得到三种能力第一理解定时器中断和回调在 MicroPython 里的正确用法第二手写一个足够简单但可扩展的协作式任务调度器第三知道在多任务场景下哪些坑是必须避开的。2. Pico 定时器与 MicroPython 多任务的基础认知2.1 Pico 的硬件定时器资源到底有多少树莓派 Pico 用的是 RP2040 芯片内置了 8 个定时器其中 4 个是通用定时器。在 MicroPython 的machine.Timer模块里Pico 暴露出来的定时器编号是 0 到 3对应这 4 个通用定时器。每个定时器可以配置为单次模式Timer.ONE_SHOT或周期模式Timer.PERIODIC支持微秒和毫秒两种单位。这里有个关键认知Pico 的定时器是硬件级别的由芯片内部的时钟驱动不占用 CPU。也就是说你在跑主循环的时候定时器计数照常进行时间到了就触发中断。这一点是整个调度器方案能成立的硬件基础。另外MicroPython 对定时器的封装非常友好两个函数就能完成配置from machine import Timer timer Timer() # 获取定时器实例 timer.init(modeTimer.PERIODIC, # 周期模式 period1000, # 每 1000 毫秒触发一次 callbacktick_handler) # 触发后执行 tick_handler2.2 MicroPython 里的“任务”到底是怎么执行的在聊调度器之前先掰扯清楚 MicroPython 的任务模型。MicroPython 本身是单线程的这意味着同一个时刻只有一个 Python 代码流在执行。但“单线程”不等于“不能多任务”关键在于如何把 CPU 时间切分给不同的任务。常见三种做法轮询式主循环里依次检查每个任务是否需要执行需要就执行。问题是如果某个任务本身耗时较长后面的任务就会饿死。定时器中断式定时器触发中断在中断回调里执行轻量级任务。问题是一旦中断回调里的代码耗时过长会影响主循环和小信号处理。协作式调度每个任务是一个短小的函数跑完就主动让出 CPU。调度器负责记录时间片到点就切下一个任务。这也是本项目的核心思路。协作式调度最优雅的地方在于不需要任务主动 yield也不需要复杂的中断栈。调度器只要靠一个“当前任务索引”和一个“时间表”就能完成基本的轮转。2.3 为什么不用 _thread 多线程很多初学者一听到多任务就往线程上想但在 Pico 上用 MicroPython 的_thread有实际痛点。首先MicroPython 的线程是基于 GIL 的同一时刻也只能跑一个 Python 线程真正并行的只有 C 扩展代码。其次线程切换是有代价的每次切换要保存和恢复上下文内存占用也会上升。最后线程之间共享变量如果不加锁很容易出现不可预期的数据竞争。对大多数 Pico 项目来说这种代价是没必要的。定时器调度器在单核环境下就能做到“任务按时间片执行”代码量少、可控性高、不容易写出玄学 bug。3. 任务调度器的设计与实现3.1 调度器的整体设计目标现在开始搭架子。我先定义好需求避免写着写着就变成“为了炫技而写复杂代码”。这个调度器要满足以下几条支持注册多个任务每个任务有自己的执行周期比如 LED 闪 0.5s 一次、按键防抖 20ms 检测一次。所有任务都在主循环的上下文里执行不在中断回调里跑任务本体。调度精度到毫秒级就够用了不追求亚毫秒。代码尽量短方便阅读和移植。基于这些需求经典做法是定时器只负责维护一个 tick 计数主循环每次都去查这个计数看哪个任务的时间片到了。3.2 第一版基于 tick 计数的轮询调度器这一版的核心逻辑很简单定时器每 10ms 触发一次中断在回调里把tick变量加一。每个任务记录一个last_tick主循环读当前tick如果和last_tick的差值超过了任务要求的周期就执行任务。上代码import machine import time TICK_MS 10 # 系统 tick 周期10ms _tick 0 _tasks [] def _tick_handler(timer): global _tick _tick 1 def add_task(func, period_ms): _tasks.append({ func: func, period: period_ms // TICK_MS, last: 0 }) def scheduler_run(): while True: current _tick for task in _tasks: if current - task[last] task[period]: task[last] current task[func]() # 没有任务要跑的时候稍微休息一下降低功耗 time.sleep_ms(1) # 示例任务 def task_led(): print(LED tick) def task_sensor(): print(Sensor read) timer machine.Timer() timer.init(modemachine.Timer.PERIODIC, periodTICK_MS, callback_tick_handler) add_task(task_led, 500) add_task(task_sensor, 100) scheduler_run()这个版本的优点是很直白任何学过 C 语言的人都能看懂。但它有个致命问题如果一个任务耗时超过了另一个任务的周期后一个任务就会被推迟执行。比如task_sensor耗时 80ms它本来 100ms 执行一次实际执行间隔却变成了 180ms。这就是协作式调度的天然特性任务必须自己控制耗时不能长时间霸占 CPU。3.3 第二版加入优先级和单次延时任务实际项目中有些任务需要尽快执行比如处理按键事件有些任务可以延后执行比如状态刷新。这时候给任务加个优先级就很有必要。另一个常见的需求是延时执行一次性的操作比如“10 秒后关掉蜂鸣器”“30 毫秒后读取传感器”。这种一次性任务如果用周期任务去实现需要手动在任务里判断状态非常繁琐。我扩展了一下数据结构_task_list [] # 周期任务 _once_list [] # 单次延时任务 _tick 0 def add_task(func, period_ms, priority0): _task_list.append({ func: func, period: period_ms // TICK_MS, last: 0, priority: priority }) _task_list.sort(keylambda t: t[priority], reverseTrue) def call_later(func, delay_ms): _once_list.append({ func: func, trigger: _tick delay_ms // TICK_MS }) def scheduler_run(): global _tick while True: current _tick # 先处理一次性任务 for task in _once_list[:]: if current task[trigger]: task[func]() _once_list.remove(task) # 再处理周期任务按优先级排序执行 for task in _task_list: if current - task[last] task[period]: task[last] current task[func]() time.sleep_ms(1)call_later是我个人非常喜欢的一个小工具写闹钟、超时控制、延时关外设非常顺手。注意这里用了一个技巧遍历_once_list时使用的是[:]拷贝这样可以在遍历的同时安全地删除元素避免迭代器失效。3.4 为什么定时器回调里不能直接执行任务这里得重点说一个非常容易踩的坑。有人觉得既然定时器中断回调能执行代码为什么不直接在回调里跑任务MicroPython 的定时器回调本质上是中断上下文的一部分。在中断上下文中大量 Python 对象操作是不安全的。比如print()虽然能输出但如果在中断里运行时间过长会让主循环无法及时响应再比如在中断回调里做列表操作append/remove有可能干扰主循环中对同一个列表的遍历轻则出现漏执行重则导致整个系统崩溃。所以本项目遵守一个原则中断里只做变量递增所有任务逻辑全部放到主循环。这个原则推广到更大规模的项目上也同样适用中断是信号灯不是办公室。4. 实战案例LED 闪烁、按键防抖、舵机扫描一起跑4.1 场景设定与接线说明为了演示调度器的实际能力我把几个常见的 Pico 外设应用塞进一个项目里让它们同时跑LED 呼吸闪烁GPIO 16 接一个 LED周期 500ms 翻转一次状态。按键防抖与状态输出GPIO 17 接一个按键用 20ms 的周期任务做防抖扫描按下时打印一次事件。舵机扫描GPIO 15 接舵机信号线让舵机在 0 度和 180 度之间来回摆动。接线很简单LED 和按键都通过面包板接到 Pico舵机直接从板子的 5V VBUS 引脚取电注意有些舵机电流大建议外接电源共地。4.2 完整代码与任务拆分import machine import time TICK_MS 10 _tick 0 _task_list [] _once_list [] def _tick_handler(timer): global _tick _tick 1 def add_task(func, period_ms, priority0): _task_list.append({ func: func, period: period_ms // TICK_MS, last: 0, priority: priority }) _task_list.sort(keylambda t: t[priority], reverseTrue) def call_later(func, delay_ms): _once_list.append({func: func, trigger: _tick delay_ms // TICK_MS}) def scheduler_run(): while True: current _tick for task in _once_list[:]: if current task[trigger]: task[func]() _once_list.remove(task) for task in _task_list: if current - task[last] task[period]: task[last] current task[func]() time.sleep_ms(1) # ---- 硬件配置 ---- led machine.Pin(16, machine.Pin.OUT) btn machine.Pin(17, machine.Pin.IN, machine.Pin.PULL_UP) servo machine.PWM(machine.Pin(15)) servo.freq(50) # ---- 任务函数 ---- led_on False def task_led_blink(): global led_on led_on not led_on led.value(1 if led_on else 0) btn_last 1 btn_state 1 def task_btn_scan(): global btn_last, btn_state # 这个函数每 20ms 跑一次 level btn.value() if level ! btn_last: btn_last level if level 0: print(Button pressed, schedule beep) call_later(play_beep, 50) def play_beep(): # 模拟蜂鸣器响应点个灯代替 led.value(0) print(Beep!) servo_angle 0 servo_dir 1 def task_servo_scan(): global servo_angle, servo_dir servo_angle servo_dir if servo_angle 180: servo_angle 180 servo_dir -1 elif servo_angle 0: servo_angle 0 servo_dir 1 # 计算 PWM 占空比 duty int(1000000 / 50 / 1000000 * 2500 servo_angle * (2500 - 500) / 180) # 简化舵机控制周期 20ms占空比 0.5ms~2.5ms duty 1966 int(servo_angle / 180 * (7864 - 1966)) servo.duty_u16(duty) # ---- 注册任务 ---- add_task(task_led_blink, 500) add_task(task_btn_scan, 20, priority10) add_task(task_servo_scan, 30) timer machine.Timer() timer.init(modemachine.Timer.PERIODIC, periodTICK_MS, callback_tick_handler) scheduler_run()这段代码跑起来后LED 以 500ms 周期翻转舵机每 30ms 改变一次角度缓缓扫描按键通过 20ms 周期扫描实现防抖按键触发后还会延迟 50ms 执行蜂鸣提示。4.3 参数的来历与计算逻辑这段代码里最容易让人困惑的是舵机 PWM 的duty_u16参数。我来拆一下计算过程舵机通常需要 50Hz 的 PWM 信号也就是周期 20ms。占空比在 0.5ms 到 2.5ms 之间对应 0° 到 180°。RP2040 的 PWM 计数器是 16 位的duty_u16的参数范围是 0 到 65535对应周期的 0% 到 100%。计算过程50Hz 时一个完整周期是 20000 微秒。0.5ms 占空比 500/20000 2.5%对应duty_u16为0.025 * 65535≈ 1638。2.5ms 占空比 2500/20000 12.5%对应duty_u16为0.125 * 65535≈ 8192。所以 0° 到 180° 的 duty 范围大约是 1638 到 8192。我在代码里写的是 1966 到 7864略作收敛避免舵机在极限位置堵转。换算成角度线性插值公式就是duty 1966 angle / 180 * (7864 - 1966)为什么不用公式int(angle * 65535 / 20000 * (2000) / 180 500/20000 * 65535)因为那样写可读性太差直接用起止范围做线性插值更直观也方便调整。4.4 任务周期的选择策略周期任务的时间片分配是有讲究的。我给三件事分配了不同的周期LED 翻转500ms 一次视觉上刚好能看到闪烁。按键扫描20ms 一次对应 50Hz 扫描频率既能保证防抖效果又不浪费 CPU。舵机扫描30ms 更新一次角度这样舵机 180° 全程走一遍大约 5 秒速度适中也不会因为更新太快导致舵机抖动。这三个周期之间没有公约数调度器靠 tick 计数来触发天然避免了同步共振问题。如果两个任务周期恰好是倍数关系它们可能总在同一时刻触发导致瞬时 CPU 压力大。错开周期比加阻塞队列更省事。5. 常见问题与排查技巧实录5.1 任务执行时间过长导致后续任务全部卡顿这是协作式调度最典型的坑。我用一个简单的办法来排查在scheduler_run里统计每个任务的执行起始时间打印耗时超过阈值的任务。def scheduler_run(): while True: current _tick for task in _task_list: if current - task[last] task[period]: task[last] current start_ms time.ticks_ms() task[func]() cost time.ticks_diff(time.ticks_ms(), start_ms) if cost 50: print(Task too slow:, task[func].__name__, cost, ms) time.sleep_ms(1)排查结果通常有两种一种是任务函数里不小心用了time.sleep()做延时阻塞了整个主循环另一种是任务里有循环等待外设响应比如 I2C 读传感器时等待数据准备好。解决办法把time.sleep()改成call_later()或改成状态机模式。比如读取传感器时可以先发起读取请求设置一个延时任务去获取结果而不是同步等数据回来。5.2 tick 溢出或回绕问题_tick是整型变量MicroPython 的整型可以无限大所以不用担心溢出。但如果把代码移植到 C 语言环境就要特别注意整数回绕问题。MicroPython 有time.ticks_ms()这样内置的处理函数专门解决了回绕比较的问题我建议在生产代码里直接用它而不是自己维护_tick。改造一下import time def task_due(last_tick, period_ms): return time.ticks_diff(time.ticks_ms(), last_tick) period_msticks_diff能正确处理回绕场景而且代码更简洁。5.3 定时器回调里的全局变量同步问题我在 3.4 提到中断回调里不要做复杂操作。但即使是递增全局变量也有一个微妙的问题MicroPython 的字节码执行过程中定时器中断可能在任意两条字节码指令之间触发。如果主循环正在读_tick回调也正好修改_tick理论上可能导致读到不一致的值。不过实际上_tick 1在 MicroPython 里是一个字节码操作而主循环读取_tick到局部变量也是一条字节码。在单核 Cortex-M0 上这种竞态在绝大多数情况下不会造成实际问题。如果你追求绝对安全可以把调度器改成基于time.ticks_ms()的重实现完全绕开自定义 tick 变量。我给的 5.2 版本代码就是这么做的建议直接参考。5.4 外设引脚冲突和 PWM 通道不足Pico 的一些引脚和板载 LED、I2C、UART 复用。比如 GPIO 25 是板载 LEDGPIO 0 和 1 默认是 UART。用 PWM 时也要注意虽然 Pico 大部分引脚都支持 PWM但同一 PWM 通道的两个引脚不能同时输出不同频率的 PWM 信号。我的建议是动手接线前先打开 Pico 的引脚功能图确认一遍把 PWM 引脚尽量分配到不同的 PWM 切片上。如果分配不合理就会出现舵机不动或 LED 亮度异常之类的诡异 bug排查起来非常费时。5.5 任务里用了阻塞式的 time.sleep 导致调度失效这个真的值得单独说。调度器跑起来之后最怕的就是某个任务函数里出现time.sleep()。这个调用会直接阻塞整个主循环让定时器的 tick 继续增加但所有任务都没机会执行。等 sleep 结束后面任务全部积压连续执行。正确做法有几种把 sleep 拆分成“先做 A延时后再做 B”的状态机。使用call_later实现延时的后半段动作。把阻塞等待改成忙等条件判断让出 CPU。哪怕是一次 100ms 的time.sleep对调度器的破坏也不容小觑。我个人的习惯是所有任务函数里不允许出现任何形式的 sleep这条规矩直接写进项目规范。5.6 任务注册顺序导致优先级失效在 3.3 的代码里我用了一个列表并且每次add_task都按优先级排序。如果忘记排序或者注册顺序和优先级冲突高优先级任务可能排在列表后面即使优先级更高也要等前面的低优先级任务跑完。这里有个更稳妥的结构用heapq或按优先级分段执行。但考虑到 MicroPython 的heapq需要额外内存我的方案是保持列表按优先级降序排列高优先级任务永远先执行。如果任务数量较多可以建立两个优先级队列高优先级列表优先遍历。5.7 定时器初始化失败或重装失败MicroPython 的 Timer 对象如果重复init()会直接重新配置硬件。但在某些固件版本中对同一个定时器实例反复 init 可能出现异常。保险做法是每个定时器只创建一次实例之后只通过deinit()再init()的方式重配置。如果遇到Timer init failed的错误优先检查是否有其他代码抢占了这个定时器。Pico 的 4 个定时器非常有限很多第三方库比如某些舵机库、超声波测距库可能会内部创建定时器和你的调度器冲突。6. 进阶扩展如何追上更复杂的调度需求6.1 从协作式调度升级到时间片轮转如果项目越来越复杂任务动不动就要 100ms 以上的执行时间协作式调度就不够用了。这时候可以考虑时间片轮转——定时器强制抢占当前任务切换到下一个任务。在 MicroPython 里实现真正的抢占式调度不现实因为语言本身不支持在任意点保存和恢复执行上下文。但在受限场景下可以做一个简化版把大任务拆成小步骤每个小步骤执行后通过call_later安排下一步。从外部看任务的流程被切碎放进了调度器效果类似时间片轮转。6.2 用状态机替代复杂任务函数状态机是嵌入式多任务的一把利器。我接手过不少复杂外设项目比如 4x4 矩阵键盘、多段温度控制曲线全部用状态机实现了“看起来像多线程”的效果。以按键长按检测为例def task_key_state_machine(): if state IDLE: if btn.value() 0: state PRESSED call_later(check_long_press, 500) elif state PRESSED: if btn.value() 1: state IDLE print(short press)这个状态机每 10ms 被调度一次但不需要任何阻塞等待长按检测靠call_later触发。这种方式比传统阻塞式按键检测可靠得多也完全不干扰其他任务。6.3 调度器的可移植性设计如果你打算把调度器从一个项目复制到另一个项目最好把代码封装成独立的模块文件scheduler.py硬件的初始化放到外部。这样在不同板子之间移植时只需要改TICK_MS和任务注册部分。Pico、ESP32、ESP8266 上帝 MicroPython 的 Timer API 略有不同比如 ESP 系列支持Timer(0)、Timer(1)这样显式创建但整体结构类似。把调度器独立出来后换平台成本极低。6.4 功耗优化与深度睡眠电池供电的项目中主循环time.sleep_ms(1)其实还是挺耗电的。更省电的做法是当没有任何任务要执行时让 CPU 进入轻睡眠直到下一次 tick 中断唤醒。MicroPython 没有直接提供轻睡眠的简单接口但可以用machine.lightsleep(1)配合定时器中断来实现。我的实测经验是加了lightsleep(1)之后正常运行时电流能降到原来的 70% 左右如果配合降低 tick 频率到 20ms整体功耗在电池项目中很可观。但这有一个风险lightsleep可能影响 USB 连接和 PWM 输出。如果调试时发现 USB 串口断开或舵机罢工先关掉睡眠优化功能稳定后再加。7. 项目总结与我的个人体会这篇东西写到这里核心的代码和思路都已经给全了。最后聊几句我自己的体会。我最开始接触 Pico 多任务时第一反应也是找现成的 RTOS研究过 FreeRTOS 的 MicroPython 移植也试过_thread。后来发现对于绝大多数个人项目和入门级原型开发一个基于定时器的协作式调度器完全够用而且因为代码是自己写的出了问题随时能改不用去啃一整个操作系统的源码。几个我坚持至今的经验定时器回调里只做最简单的计数任何任务逻辑都回到主循环执行。任务函数永远不要用time.sleep()要延时就用call_later。给所有任务设置合理的周期不要追求 1ms 级精度MicroPython 处理不了这么高频的 Python 函数调用你也不需要这种精度。状态机加调度器的组合能处理 90% 以上的复杂外设逻辑代码还特别好调试。后续想再折腾的话可以试着把调度器加上任务时间统计、任务暂停和恢复、动态注册任务这些功能。对 Pico 项目来说这套框架扩展到几十个任务也没问题前提是每个任务自己别偷懒霸占 CPU。还是那句话先把基础版本跑通再谈优化。定时器驱动的任务调度器真的没有想象中那么高不可攀。

读完文章,也想定制专属网站?

尧图设计师 24 小时内与您沟通定制方案

免费获取报价