注册有礼

用apscheduler模块怎样实现定时任务，操作是什么

Admin 2022-08-19 群英技术资讯 504 次浏览

上一篇：opencv颜色空间转换的方法是什么，怎样实现 下一篇：Python序列化pickle和json怎么用，两者有何差异
这篇文章主要讲解了“用apscheduler模块怎样实现定时任务，操作是什么”，文中的讲解内容简单、清晰、详细，对大家学习或是工作可能会有一定的帮助，希望大家阅读完这篇文章能有所收获。下面就请大家跟着小编的思路一起来学习一下吧。


 
一、安装
  

pip install apscheduler

  

 

二、ApScheduler 简介
 

 

1 APScheduler的组件
 

triggers：触发器
triggers包含任务执行的调度逻辑，决定任务按照什么逻辑进行定时执行
 

job stores；任务存储器
存储了调度任务
 

executors：执行器
用例执行任务的，包含线程池以及进程池等的创建和调用等等
 

schedulers：调度器
属于控制面，将其他几个方面组织起来的作用、
 

 

2 调度器的种类
 

调度器有以下几种常见类型，其中最常用的BackgroundScheduler，即非阻塞式，因为在一般情况下，定时任务都会在放到web服务中，如果使用阻塞式，则无法启动web服务，而使用非阻塞式，则将定时任务设定后，就不管了，继续执行后面的web服务，只要web服务在运行，定时任务就是一直有效的
 

     
 
  • BlockingScheduler: 阻塞式
    •  
 
  • BackgroundScheduler: 非阻塞式（后台运行）
    •  
 
  • AsyncIOScheduler: 当使用asyncio模块时使用
    •  
 
  • GeventScheduler: 当使用gevent模块时使用
    •  
 
  • TornadoScheduler: 构建Tornado应用时使用
    •  
 
  • TwistedScheduler: 构建Twisted应用时使用
    •  
 
  • QtScheduler: 构建Qt应用时使用
    •  

 

 

3 内置的触发器类型
 

     
 
  • date: 在某个时间点执行一次时使用
    •  
 
  • interval: 固定的时间间隔循环执行时使用
    •  
 
  • cron: 在一天中特定的时间点执行时使用
    •  
 
  • calendarinterval: 当想在在一天中特定时间点或以日历为基础的时间间隔内执行时使用
    •  

 

 

三、使用举例
 

这里jiu就以非阻塞式BackgroundScheduler调度器为例展开
 

 

1 使用date类型的触发器
 

如下，使用了三种设置日期和时间的方法
  

from apscheduler.schedulers.background import BackgroundScheduler
import time
from datetime import date
from datetime import datetime


def do_func(name,age):
    print(time.strftime('%Y-%m-%d %H:%M:%S',time.localtime(time.time()))+" in do func : 姓名："+name+" 年龄："+str(age))

def main():
    print(time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time())))
    sched=BackgroundScheduler()

    # 通过date 设置指定日期执行
    sched.add_job(do_func,trigger="date",run_date=date(2022,5,25),args=("张三丰",100))

    # 通过datetime，设置指定日期额指定时刻执行
    sched.add_job(do_func, trigger="date", run_date=datetime(2022, 5, 25,14,0,10), args=("张三丰", 100))

    # 直接使用文本的方式指定日期和时刻表
    sched.add_job(do_func, trigger="date", run_date="2022-05-25 14:0:20", args=("张三丰", 100))

    sched.start()

if __name__=="__main__":
    main()
    while True:
        print(time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time())))
        time.sleep(1)
  

执行结果如下,可以发现，第一个通过date指定日期的默认是0点执行，显然时间已经过了，不会执行，第二和第三个则在规定的时间点执行了，这里还需要注意的是，通过打印可以看出，main函数执行完成后，已经开始执行main函数下面的while循环打印语句了，而在执行循环的过程中，定时任务仍然生效，这就是非阻塞式调度器的原理，如果是阻塞式，则在此代码中，会一直卡在main函数中，main下面额while循环语句是不会执行的，因此在实际使用中，非阻塞式应用的是非常多的
 

 
 
2022-05-25 14:00:02
2022-05-25 14:00:02
C:\Users\hitre\.virtualenvs\zentaolinkgitlab-QCS5yxD9\lib\site-packages\apscheduler\util.py:436: PytzUsageWarning: The localize method is no longer necessary, as this time zone supports the fold attribute (PEP 495). For more details on migrating to a PEP 495-compliant implementation, see https://pytz-deprecation-shim.readthedocs.io/en/latest/migration.html
  return tzinfo.localize(dt)
Run time of job "do_func (trigger: date[2022-05-25 00:00:00 CST], next run at: 2022-05-25 00:00:00 CST)" was missed by 14:00:02.088260
2022-05-25 14:00:03
2022-05-25 14:00:04
2022-05-25 14:00:05
2022-05-25 14:00:06
2022-05-25 14:00:07
2022-05-25 14:00:08
2022-05-25 14:00:09
2022-05-25 14:00:10 in do func : 姓名：张三丰 年龄：100
2022-05-25 14:00:10
2022-05-25 14:00:11
2022-05-25 14:00:12
2022-05-25 14:00:13
2022-05-25 14:00:14
2022-05-25 14:00:15
2022-05-25 14:00:16
2022-05-25 14:00:17
2022-05-25 14:00:18
2022-05-25 14:00:19
2022-05-25 14:00:20 in do func : 姓名：张三丰 年龄：100
2022-05-25 14:00:20
2022-05-25 14:00:21
2022-05-25 14:00:22
 

 

 

2 使用interval类型的触发器
 

如下代码演示了时间间隔循环执行的使用例子
  

from apscheduler.schedulers.background import BackgroundScheduler
import time
from datetime import date
from datetime import datetime


def do_func(name,age):
    print(time.strftime('%Y-%m-%d %H:%M:%S',time.localtime(time.time()))+" in do func : 姓名："+name+" 年龄："+str(age))

def main():
    print(time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time())))
    sched=BackgroundScheduler()

    # 每3秒执行一次
    sched.add_job(do_func,trigger="interval",args=("张三丰",100),seconds=3)
    # 每3分钟执行一次
    sched.add_job(do_func, trigger="interval", args=("张三丰", 100), minutes=3)
    # 每3小时执行一次
    sched.add_job(do_func, trigger="interval", args=("张三丰", 100), hours=3)
    # 每3天执行一次
    sched.add_job(do_func, trigger="interval", args=("张三丰", 100), days=3)
    # 每3周执行一次
    sched.add_job(do_func, trigger="interval", args=("张三丰", 100), weeks=3)

    sched.start()

if __name__=="__main__":
    main()
    while True:
        print(time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time())))
        time.sleep(1)
  

上面的代码中因为时间跨度比较大，这里只演示妹3秒执行一次的代码
 

代码如下：
  

from apscheduler.schedulers.background import BackgroundScheduler
import time
from datetime import date
from datetime import datetime


def do_func(name,age):
    print(time.strftime('%Y-%m-%d %H:%M:%S',time.localtime(time.time()))+" in do func : 姓名："+name+" 年龄："+str(age))

def main():
    print(time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time())))
    sched=BackgroundScheduler()

    # 每3秒执行一次
    sched.add_job(do_func,trigger="interval",args=("张三丰",100),seconds=3)

    sched.start()

if __name__=="__main__":
    main()
    while True:
        print(time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time())))
        time.sleep(1)
  

执行结果如下：
 

 
 
2022-05-25 14:14:04
2022-05-25 14:14:04
2022-05-25 14:14:05
2022-05-25 14:14:06
2022-05-25 14:14:07 in do func : 姓名：张三丰 年龄：100
2022-05-25 14:14:07
2022-05-25 14:14:08
2022-05-25 14:14:09
2022-05-25 14:14:10 in do func : 姓名：张三丰 年龄：100
2022-05-25 14:14:10
2022-05-25 14:14:11
2022-05-25 14:14:12
2022-05-25 14:14:13 in do func : 姓名：张三丰 年龄：100
2022-05-25 14:14:13
 

 

 

3 使用cron类型的触发器
 

cron的触发器有点类似linux上crontab定时器的使用，代码如下：
  

from apscheduler.schedulers.background import BackgroundScheduler
import time
from datetime import date
from datetime import datetime


def do_func(name,age):
    print(time.strftime('%Y-%m-%d %H:%M:%S',time.localtime(time.time()))+" in do func : 姓名："+name+" 年龄："+str(age))

def main():
    print(time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time())))
    sched=BackgroundScheduler()

    # 任务会在6月、7月、8月、11月和12月的第三个周五，00:00、01:00、02:00和03:00触发
    sched.add_job(do_func,trigger="cron",month='6-8,11-12', day='3rd fri', hour='0-3',args=("张三丰",100))

    sched.start()

if __name__=="__main__":
    main()
    while True:
        print(time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time())))
        time.sleep(1)
  

这里需要注意的是，可以省略不需要的字段。当省略时间参数时，在显式指定参数之前的参数会被设定为*，之后的参数会被设定为最小值，week 和day_of_week的最小值为*
  

day=1, minute=20
等同于
year='*', month='*', day=1, week='*', day_of_week='*', hour='*', minute=20, second=0
  

此外，也可以直接使用crontab表达式，如下：
  

from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.triggers.cron import CronTrigger
import time
from datetime import date
from datetime import datetime


def do_func(name,age):
    print(time.strftime('%Y-%m-%d %H:%M:%S',time.localtime(time.time()))+" in do func : 姓名："+name+" 年龄："+str(age))

def main():
    print(time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time())))
    sched=BackgroundScheduler()

    # 任务会在6月、7月、8月、11月和12月的第三个周五，00:00、01:00、02:00和03:00触发
    sched.add_job(do_func,trigger=CronTrigger.from_crontab('48 10 1-15 sep-nov *'),args=("张三丰",100))

    sched.start()

if __name__=="__main__":
    main()
    while True:
        print(time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time())))
        time.sleep(1)
  

 

四、定时器使用装饰器的方法
 

以间隔时间循环执行的代码为例，如下为未使用装饰器的方式
  

from apscheduler.schedulers.background import BackgroundScheduler
import time
from datetime import date
from datetime import datetime


def do_func(name,age):
    print(time.strftime('%Y-%m-%d %H:%M:%S',time.localtime(time.time()))+" in do func : 姓名："+name+" 年龄："+str(age))

def main():
    print(time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time())))
    sched=BackgroundScheduler()

    # 每3秒执行一次
    sched.add_job(do_func,trigger="interval",args=("张三丰",100),seconds=3)

    sched.start()

if __name__=="__main__":
    main()
    while True:
        print(time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time())))
        time.sleep(1)
  

修改为使用装饰器的方式如下：
  

from apscheduler.schedulers.background import BackgroundScheduler
import time

sched=BackgroundScheduler()

@sched.scheduled_job(trigger="interval",args=("张三丰",100),seconds=3)
def do_func(name,age):
    print(time.strftime('%Y-%m-%d %H:%M:%S',time.localtime(time.time()))+" in do func : 姓名："+name+" 年龄："+str(age))

if __name__=="__main__":
    sched.start()
    while True:
        print(time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time())))
        time.sleep(1)
  

执行结果如下：
 

 
 
2022-05-25 14:34:10
2022-05-25 14:34:11
2022-05-25 14:34:12
2022-05-25 14:34:13 in do func : 姓名：张三丰 年龄：100
2022-05-25 14:34:13
2022-05-25 14:34:14
2022-05-25 14:34:15
2022-05-25 14:34:16 in do func : 姓名：张三丰 年龄：100
2022-05-25 14:34:16
2022-05-25 14:34:17
2022-05-25 14:34:18
2022-05-25 14:34:19 in do func : 姓名：张三丰 年龄：100
2022-05-25 14:34:19
2022-05-25 14:34:20
     


以上就是关于“用apscheduler模块怎样实现定时任务，操作是什么”的介绍了，感谢各位的阅读，如果大家想要了解更多相关的内容，欢迎关注群英网络，小编每天都会为大家更新不同的知识。
                
                                
                    群英智防CDN，智能加速解决方案
                
                                
                                

                    标签：
                                           apscheduler模块
                                    

                
                
免责声明：本站发布的内容（图片、视频和文字）以原创、转载和分享为主，文章观点不代表本网站立场，如果涉及侵权请联系站长邮箱：mmqy2019@163.com进行举报，并提供相关证据，查实之后，将立刻删除涉嫌侵权内容。

                
                

                                            上一篇：opencv颜色空间转换的方法是什么，怎样实现
                                                                下一篇：Python序列化pickle和json怎么用，两者有何差异  
                                    
                

                
猜你喜欢

                
            

            

                
                    云活动
                
                
                    云活动
                
                
                

                    
推荐内容

                    

                

                

                    
相关标签

                    

                

                

            


        


    


 
    

        

		

			
成为群英会员,开启智能安全云计算之旅

			
			立即注册
		

	



	

		

			

				 专业资深工程师驻守
			

			

				 7X24小时快速响应
			

			

				 一站式无忧技术支持
			

			

				 免费备案服务
			

		

	





	

		
	
			

				

					产品与服务
					
						云服务器
					
					
						智防系统
					
					
						服务器租用
					
					
						服务器托管
					
					
					
					
						虚拟主机
					

				

				

					帮助中心 注册与购买
					
											
							备案问题
						
						
							鹰云服务器问题
						
						
							服务器托管问题
						
						
							域名问题
						
											
							高防服务器问题
						

						
							服务器租用问题
						

				

				

					关于群英 
					
						群英简介
					
										
						公司动态
					
					
						资质荣誉
					
					 
						 发展历程
						
					 
						 群英环境
						
					
						汇款信息
					 
					
						备案中心
					
										
						联系我们
					
				

				

					其他 
					
						数据中心
					
					
						生态合作
					
				

				

					
						友情链接 

						友链合作
					 
					
					
						群英网络
						
						智帝云
					
					
						主机评测
						
						idc公司
					
					
						商城系统
						
						跨境电商
					
					
						主机吧
						
						酷库博客
					
					
						OA系统
						
						腾讯企业邮箱
					
					
						锐成信息
						
						sdwan
					
				


				

					
关注或联系群英网络

					

						7x24小时售前：400-678-4567
					

					

						7x24小时售后：0668-2555666
					

					

						
						
24小时QQ客服
	
					

					

						
						
群英微信公众号
	
					

				

			

		
		

	

		

		
		

			Copyright © QY Network Company Ltd. All Rights Reserved. 2003-2020 群英 版权所有 
		

		

			增值电信经营许可证 : B1.B2-20140078   
				粤ICP备09006778号 
				域名注册商资质 粤 D3.1-20240008
				
		

		
		

			 
		


	

		


    

    

		 免费拨打  400-678-4567
				

					免费拨打  400-678-4567 免费拨打 400-678-4567 或
							0668-2555555
				
				

		

			在线客服
				

					
				

		
		

		 在线客服
				
  微信公众号
				
  返回顶部
				

					返回顶部 返回顶部
				
				

	

	

		在线客服
	
	

	

		在线客服