『micropython』的任务调度器

allankliu

『micropython』的任务调度器 [复制链接]

 

前一阶段看到有网友希望在microPython上运行多线程。我觉得这个有些困难。首先，CPython就没有解决GIL的问题。所以多线程就是一个梦。Java是可以的。但是任务调度的确是一个需求。

从Python2.7开始，标准库里就有sched。但是micropython没有实现。不过micropython的作者推荐了一个简单的调度器类实现。

1. #A demo scheduler.py
   
2. 
   
3. import pyb
   
4. 
   
5. class Scheduler:
   
6.     def __init__(self):
   
7.         self.events = []
   
8. 
   
9.     def schedule(self, delay, function):
   
10.         if delay <= 0:
    
11.             function()
    
12.         else:
    
13.             self.events.append([delay, function])
    
14.             self.events.sort(key=lambda event:event[0])
    
15. 
    
16.     def run(self):
    
17.         while len(self.events) > 0:
    
18.             delay, function = self.events.pop(0)
    
19.             for event in self. events:
    
20.                 event[0] -= delay
    
21.             pyb.delay(delay)
    
22.             function()
    
23.             
    
24. class LED:
    
25.     def __init__(self, led_id, time_on, time_off):
    
26.         self.led_id = led_id
    
27.         self.time_on = time_on
    
28.         self.time_off = time_off
    
29.         self.on()
    
30.         
    
31.     def on(self):
    
32.         pyb.LED(self.led_id, True)
    
33.         sched.schedule(self.time_on, self.off)
    
34.         
    
35.     def off(self):
    
36.         pyb.LED(self.led_id, False)
    
37.         sched.schedule(self.time_off, self.on)
    
38.         
    
39. sched = Scheduler()
    
40. LED(1, 1000, 2000)
    
41. LED(2, 3000, 1000)
    
42. sched.run()

复制代码

但是作者提供的代码有语法错误。（提供大家纠错）我将代码稍微改成另外的形式。

1. #A demo scheduler.py
   
2. 
   
3. import pyb
   
4. 
   
5. class Scheduler:
   
6.     def __init__(self):
   
7.         self.events = []
   
8. 
   
9.     def schedule(self, delay, function):
   
10.         if delay <= 0:
    
11.             function()
    
12.         else:
    
13.             self.events.append([delay, function])
    
14.             self.events.sort(key=lambda event:event[0])
    
15. 
    
16.     def run(self):
    
17.         while len(self.events) > 0:
    
18.             delay, function = self.events.pop(0)
    
19.             for event in self. events:
    
20.                 event[0] -= delay
    
21.             pyb.delay(delay)
    
22.             function()
    
23. 
    
24. def job1():
    
25.     pyb.LED(1).toggle()
    
26.     sched.schedule(1000,job2)
    
27.    
    
28. def job2():
    
29.     pyb.LED(2).toggle()
    
30.     sched.schedule(2000,job1)
    
31.         
    
32. sched = Scheduler()
    
33. job1()
    
34. job2()
    
35. sched.run()

复制代码

但是Scheduler类中，时间调度采用pyb.delay()来使用，是阻塞型的调度器。不能够让人满意。不过，暂时可以提供给大家使用吧。

okhxyyo

汇总在此：
【MicroPython】——by allankliu
https://bbs.eeworld.com.cn/forum ... 2739&fromuid=536508

nmg

这个网友好像就是@windworld

dcexpert

太厉害了。

Onsunsl

类似一个伪调度器

allankliu

堵塞型的东西，核心思想是利用List的queue队列特性。如果要做成时间切片或者优先级调度，需要更加复杂的设计。

前者需要利用单独定时器来做调度队列中的任务，后者则需要在任务列表容器中增加优先级，pop/push任务时可以进行排序。

基于python的数据结构，实现起来并不算难。

gmyu

LZ继续