首页 > 代码库 > [Dynamic Language] Python定时任务框架

[Dynamic Language] Python定时任务框架

APScheduler是一个Python定时任务框架,使用起来十分方便。提供了基于日期、固定时间间隔以及crontab类型的任务,并且可以持久化任务、并以daemon方式运行应用。

测试计划任务

mac-abeen:timetask abeen$ vim testtask.py

   1 # !/usr/bin/env python                                                                                                                                                                                    2 # -*- encoding:utf-8 -*-                                                                  3                                                                                           4 from datetime import datetime                                                             5                                                                                           6                                                                                           7 class TestTask(object):                                                                   8     """                                                                                   9         测试计划任务                                                                     10     """                                                                                  11                                                                                          12     def print_file(self, content):                                                       13         f = open("testtask.log", "a")                                                    14         f.write(content)                                                                 15         f.close()                                                                        16                                                                                          17                                                                                          18     def run_second(self):                                                                19         self.print_file("run_second is run on {0} \r\n ".format(datetime.now()))         20                                                                                          21                                                                                          22     def run_minute(self):                                                                23         self.print_file("run_minute is run on {0} \r\n ".format(datetime.now()))         24                                                                                          25                                                                                          26                                                                                          27                                                                                          28 if __name__ == "__main__":                                                               29     tt = TestTask()                                                                      30     tt.run_second()                                                                      31     tt.run_minute()  

 

任务执行
mac-abeen:timetask abeen$ vim run_task.py

   1 # !/usr/bin/env python                                                                                                                                                                                    2 # -*- encoding:utf-8 -*-                                                                  3                                                                                           4 from datetime import datetime                                                             5 import time                                                                               6 from apscheduler.schedulers.blocking import BlockingScheduler                             7 from testtask import TestTask                                                             8                                                                                           9 def run():                                                                               10     scheduler = BlockingScheduler()                                                      11     scheduler.add_job(func=TestTask().run_second, trigger=‘interval‘, seconds=3)         12     scheduler.add_job(func=TestTask().run_minute, trigger=‘interval‘, seconds=60)        13                                                                                          14     scheduler.start()                                                                    15                                                                                          16                                                                                          17 if __name__ == "__main__":                                                               18     print "scheduler is run ......"                                                      19     run() 

 日志记录(测试部分日志信息)

1 run_second is run on 2017-06-29 15:13:16.070406                                                                                                                                                           2  run_minute is run on 2017-06-29 15:13:16.071498   3  run_second is run on 2017-06-29 15:13:32.378192   ...  22  run_minute is run on 2017-06-29 15:14:29.379333  23  run_second is run on 2017-06-29 15:14:29.379462  24  run_second is run on 2017-06-29 15:14:32.380961 ...  44  run_minute is run on 2017-06-29 15:15:29.380135  45  run_second is run on 2017-06-29 15:15:32.380093  46  run_second is run on 2017-06-29 15:15:35.378075....

 

 


官方资料帮助:
https://pypi.python.org/pypi/APScheduler/#downloads

[Dynamic Language] Python定时任务框架