Python使用apscheduler模块设置定时任务的实现
作者:redrose2100 发布时间:2021-05-01 05:01:54
一、安装
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
来源:https://blog.csdn.net/redrose2100/article/details/124965778
![](https://www.aspxhome.com/images/zang.png)
![](https://www.aspxhome.com/images/jiucuo.png)
猜你喜欢
- 以下摘录自 oreilly.JavaScript.The.Definitive.Guide.5th.Edition.Aug.200
- 游标(Cursor)是处理数据的一种方法,为了查看或者处理结果集中的数据,游标提供了在结果集中一次以行或者多行前进或向后浏览数据的能力。我们
- 一、概论超大型系统的特点为:1、处理的用户数一般都超过百万,有的还超过千万,数据库的数据量一般超过1TB;2、系统必须提供实时响应功能,系统
- CONVERT将某种数据类型的表达式显式转换为另一种数据类型。由于某些需求经常用到取日期格式的不同.现以下可在SQL Server中将日期格
- 问题你有50枚金币,需要分配给以下几个人:Matthew,Sarah,Augustus,Heidi,Emilie,Peter,Giana,A
- Yahoo和Google都有自己的建设高性能网站最佳实践, 我不做赘述, 需要了解的自行查阅资料:Yahoo的: Best Practice
- 什么是 conda ?conda 是开源包(packages)和虚拟环境(environment)的管理系统。**packages 管理:*
- 长话短说,今天介绍实现此功能的一个方法,需要了解的朋友可以参考下:一、JS 重载页面,本地刷新,返回上一页 代码如下:<a href=
- 标准库的fnmatch库专门用来进行文件名匹配,支持使用通配符进行字符串匹配。1、fnmatch:判断文件名是否符合特定的模式;2、fnma
- 很简单的教程,献给喜欢SEO的朋友们。把article.asp?logID=26 替换成article.asp?/a
- python基于新浪sae开发的微信公众平台,实现功能:输入段子---回复笑话输入开源+文章---发送消息到开源中国输入快递+订单号---查
- 以去掉去掉<!--和-->为例进行说明:def clearContentWithSpecialCharacter(content
- 前言回调函数是我们在python编程中经常会遇到的一个问题,而想在将来某一时刻进行函数回调,可以使用call_later()函数来实现,第一
- 目录需求分析进一步分析再进一步分析代码实现我们在写爬虫的过程中,除了研究反爬之外,几乎全部的时间都在写解析逻辑。那么,生命苦短,为什么我们不
- 由于我们分发的python应用可能运行在64位环境,也可能运行在32位环境,所以我们需要为同一套应用代码配置两套打包环境,怎么配置?步骤如下
- Python产生一个数值范围内的不重复的随机数,可以使用random模块中的random.sample函数,其用法如下:import ran
- LyScript 插件默认提供了一个get_disasm_code()方法可以直接获取到指定行数的反汇编代码,但如果需要自定义获取或者是需要
- 当遇到一个模板标签(template tag)时,模板解析器就会把标签包含的内容,以及模板解析器自己作为参数调用一个python函数。 这个
- 1.解读tensorflow权重文件,透过 tf.train.NewCheckpointReader函数。2.reader.get_vari
- 一). ubuntu下mysql安装布局:/usr/bin