Python的sched模块和Timer类

我们经常需要定时的执行某个任务,在Linux下我们有强大的crontab,但是在Python这个粒度(定时执行函数),如何处理呢?
除了第三方的模块外,标准库为我们提供了sched模块和Timer类。

先说sched模块,准确的说,它是一个调度(延时处理机制),每次想要定时执行某任务都必须写入一个调度。
使用步骤如下:
(1)生成调度器:
s = sched.scheduler(time.time,time.sleep)
第一个参数是一个可以返回时间戳的函数,第二个参数可以在定时未到达之前阻塞。可以说sched模块设计者是“在下很大的一盘棋”,比如第一个函数可以是自定义的一个函数,不一定是时间戳,第二个也可以是阻塞socket等。
(2)加入调度事件
其实有enter、enterabs等等,我们以enter为例子。
s.enter(x1,x2,x3,x4)
四个参数分别为:间隔事件、优先级(用于同时间到达的两个事件同时执行时定序)、被调用触发的函数,给他的参数(注意:一定要以tuple给如,如果只有一个参数就(xx,))
(3)运行
s.run()
注意sched模块不是循环的,一次调度被执行后就Over了,如果想再执行,请再次enter

Plain text
Copy to clipboard
Open code in new window
EnlighterJS 3 Syntax Highlighter
import time,sched
#被调度触发的函数
def event_func(msg):
print "Current Time:",time.time(),'msg:',msg
if __name__ == "__main__":
#初始化sched模块的scheduler类
s = sched.scheduler(time.time,time.sleep)
#设置两个调度
s.enter(1,2,event_func,("Small event.",))
s.enter(2,1,event_func,("Big event.",))
s.run()
while True:
time.sleep(100)
import time,sched #被调度触发的函数 def event_func(msg): print "Current Time:",time.time(),'msg:',msg if __name__ == "__main__": #初始化sched模块的scheduler类 s = sched.scheduler(time.time,time.sleep) #设置两个调度 s.enter(1,2,event_func,("Small event.",)) s.enter(2,1,event_func,("Big event.",)) s.run() while True: time.sleep(100)
import time,sched

#被调度触发的函数
def event_func(msg):
    print "Current Time:",time.time(),'msg:',msg

if __name__ == "__main__":
    #初始化sched模块的scheduler类
    s = sched.scheduler(time.time,time.sleep)
    #设置两个调度
    s.enter(1,2,event_func,("Small event.",))
    s.enter(2,1,event_func,("Big event.",))
    s.run()
    while True:
        time.sleep(100)

Timer类则是居然也不是循环使用的,但它的灵活性要差很多,用法也比较简单。

Plain text
Copy to clipboard
Open code in new window
EnlighterJS 3 Syntax Highlighter
import threading
import time
def test_func(msg1,msg2):
print "I'm test_func,",msg1,msg2
if __name__ == "__main__":
t = threading.Timer(5,test_func,("msg1","msg2"))
t.start()
while True:
time.sleep(1)
import threading import time def test_func(msg1,msg2): print "I'm test_func,",msg1,msg2 if __name__ == "__main__": t = threading.Timer(5,test_func,("msg1","msg2")) t.start() while True: time.sleep(1)
import threading
import time

def test_func(msg1,msg2):
    print "I'm test_func,",msg1,msg2

if __name__ == "__main__":
    t = threading.Timer(5,test_func,("msg1","msg2"))
    t.start()
    while True:
        time.sleep(1)

如何让Timer循环呢,一种很山寨的方法是……在test_func中再次调用Timer...start...

Plain text
Copy to clipboard
Open code in new window
EnlighterJS 3 Syntax Highlighter
import threading
import time
def timer_start():
t = threading.Timer(5,test_func,("msg1","msg2"))
t.start()
def test_func(msg1,msg2):
print "I'm test_func,",msg1,msg2
timer_start()
if __name__ == "__main__":
timer_start()
while True:
time.sleep(1)
import threading import time def timer_start(): t = threading.Timer(5,test_func,("msg1","msg2")) t.start() def test_func(msg1,msg2): print "I'm test_func,",msg1,msg2 timer_start() if __name__ == "__main__": timer_start() while True: time.sleep(1)
import threading
import time

def timer_start():
    t = threading.Timer(5,test_func,("msg1","msg2"))
    t.start()

def test_func(msg1,msg2):
    print "I'm test_func,",msg1,msg2
    timer_start()

if __name__ == "__main__":
    timer_start()
    while True:
        time.sleep(1)

我再研究一下有没有更优雅的实现……哎。。

3 thoughts on “Python的sched模块和Timer类

  1. fireflyliu

    最近刚好需要使用这个模块,事前我居然不知道sched这个内置的模块,看了老兄的文章,醍醐灌顶啊,谢谢了

    Reply
  2. Kyle Sawyer

    其实sched是可以循环的,比如这样来写:

    scheduler = sched.scheduler(time.time, time.sleep)

    def event(action_time):
    print "action_time:%s" % (action_time)
    scheduler.enterabs(action_time + 5, 1, event, (action_time + 5,))

    inittime = time.time()

    scheduler.enterabs(inittime, 1, event, (inittime,))

    Reply
  3. Anonymous

    时隔4年。。。居然在这会搜索到跟我一样,用同样的方法实现循环Timer的。。。。

    Reply

Leave a Reply

Your email address will not be published. Required fields are marked *