[Dynamic Language] Python定时任务框架

Posted 2020-09-25

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