Python使用APScheduler实现定时任务过程解析
作者:gdjlc 发布时间:2023-01-23 19:20:35
前言
APScheduler是基于Quartz的一个Python定时任务框架。提供了基于日期、固定时间间隔以及crontab类型的任务,并且可以持久化任务。
在线文档:https://apscheduler.readthedocs.io/en/latest/userguide.html
一、安装APScheduler
pip install apscheduler
二、基本概念
APScheduler有四大组件:
1、触发器 triggers :
触发器包含调度逻辑。每个作业都有自己的触发器,用于确定下一个任务何时运行。除了初始配置之外,触发器是完全无状态的。
有三种内建的trigger:
(1)date: 特定的时间点触发
(2)interval: 固定时间间隔触发
(3)cron: 在特定时间周期性地触发
2、任务储存器 job stores:用于存放任务,把任务存放在内存(为默认MemoryJobStore)或数据库中。
3、执行器 executors: 执行器是将任务提交到线程池或进程池中运行,当任务完成时,执行器通知调度器触发相应的事件。
4、调度器 schedulers: 把上方三个组件作为参数,通过创建调度器实例来运行
根据开发需求选择相应的组件,下面是不同的调度器组件:
BlockingScheduler 阻塞式调度器:适用于只跑调度器的程序。
BackgroundScheduler 后台调度器:适用于非阻塞的情况,调度器会在后 * 立运行。
AsyncIOScheduler AsyncIO调度器,适用于应用使用AsnycIO的情况。
GeventScheduler Gevent调度器,适用于应用通过Gevent的情况。
TornadoScheduler Tornado调度器,适用于构建Tornado应用。
TwistedScheduler Twisted调度器,适用于构建Twisted应用。
QtScheduler Qt调度器,适用于构建Qt应用。
三、使用步骤
1、新建一个调度器schedulers
2、添加调度任务
3、运行调度任务
四、使用实例
1、触发器date
特定的时间点触发,只执行一次。参数如下:
参数 | 说明 |
run_date (datetime 或 str) | 作业的运行日期或时间 |
timezone (datetime.tzinfo 或 str) | 指定时区 |
使用例子:
from datetime import datetime
from datetime import date
from apscheduler.schedulers.blocking import BlockingScheduler
def job(text):
print(text)
scheduler = BlockingScheduler()
# 在 2019-8-30 运行一次 job 方法
scheduler.add_job(job, 'date', run_date=date(2019, 8, 30), args=['text1'])
# 在 2019-8-30 01:00:00 运行一次 job 方法
scheduler.add_job(job, 'date', run_date=datetime(2019, 8, 30, 1, 0, 0), args=['text2'])
# 在 2019-8-30 01:00:01 运行一次 job 方法
scheduler.add_job(job, 'date', run_date='2019-8-30 01:00:00', args=['text3'])
scheduler.start()
2、触发器interval
固定时间间隔触发。参数如下:
参数 | 说明 |
weeks (int) | 间隔几周 |
days (int) | 间隔几天 |
hours (int) | 间隔几小时 |
minutes (int) | 间隔几分钟 |
seconds (int) | 间隔多少秒 |
start_date (datetime 或 str) | 开始日期 |
end_date (datetime 或 str) | 结束日期 |
timezone (datetime.tzinfo 或str) |
使用例子:
import time
from apscheduler.schedulers.blocking import BlockingScheduler
def job(text):
t = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time()))
print('{} --- {}'.format(text, t))
scheduler = BlockingScheduler()
# 每隔 1分钟 运行一次 job 方法
scheduler.add_job(job, 'interval', minutes=1, args=['job1'])
# 在 2019-08-29 22:15:00至2019-08-29 22:17:00期间,每隔1分30秒 运行一次 job 方法
scheduler.add_job(job, 'interval', minutes=1, seconds = 30, start_date='2019-08-29 22:15:00', end_date='2019-08-29 22:17:00', args=['job2'])
scheduler.start()
'''
运行结果:
job2 --- 2019-08-29 22:15:00
job1 --- 2019-08-29 22:15:46
job2 --- 2019-08-29 22:16:30
job1 --- 2019-08-29 22:16:46
job1 --- 2019-08-29 22:17:46
...余下省略...
'''
3、触发器cron
在特定时间周期性地触发。参数如下:
参数 | 说明 |
year (int 或 str) | 年,4位数字 |
month (int 或 str) | 月 (范围1-12) |
day (int 或 str) | 日 (范围1-31) |
week (int 或 str) | 周 (范围1-53) |
day_of_week (int 或 str) | 周内第几天或者星期几 (范围0-6 或者 mon,tue,wed,thu,fri,sat,sun) |
hour (int 或 str) | 时 (范围0-23) |
minute (int 或 str) | 分 (范围0-59) |
second (int 或 str) | 秒 (范围0-59) |
start_date (datetime 或 str) | 最早开始日期(包含) |
end_date (datetime 或 str) | 最晚结束时间(包含) |
timezone (datetime.tzinfo 或str) | 指定时区 |
这些参数支持算数表达式,取值格式有如下:
使用例子:
import time
from apscheduler.schedulers.blocking import BlockingScheduler
def job(text):
t = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time()))
print('{} --- {}'.format(text, t))
scheduler = BlockingScheduler()
# 在每天22点,每隔 1分钟 运行一次 job 方法
scheduler.add_job(job, 'cron', hour=22, minute='*/1', args=['job1'])
# 在每天22和23点的25分,运行一次 job 方法
scheduler.add_job(job, 'cron', hour='22-23', minute='25', args=['job2'])
scheduler.start()
'''
运行结果:
job1 --- 2019-08-29 22:25:00
job2 --- 2019-08-29 22:25:00
job1 --- 2019-08-29 22:26:00
job1 --- 2019-08-29 22:27:00
...余下省略...
'''
4、通过装饰器scheduled_job()添加方法
添加任务的方法有两种:
(1)通过调用add_job()---见上面1至3代码
(2)通过装饰器scheduled_job():
第一种方法是最常用的方法。第二种方法主要是方便地声明在应用程序运行时不会更改的任务。该 add_job()方法返回一个apscheduler.job.Job实例,可以使用该实例稍后修改或删除该任务。
import time
from apscheduler.schedulers.blocking import BlockingScheduler
scheduler = BlockingScheduler()
@scheduler.scheduled_job('interval', seconds=5)
def job1():
t = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time()))
print('job1 --- {}'.format(t))
@scheduler.scheduled_job('cron', second='*/7')
def job2():
t = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time()))
print('job2 --- {}'.format(t))
scheduler.start()
'''
运行结果:
job2 --- 2019-08-29 22:36:35
job1 --- 2019-08-29 22:36:37
job2 --- 2019-08-29 22:36:42
job1 --- 2019-08-29 22:36:42
job1 --- 2019-08-29 22:36:47
job2 --- 2019-08-29 22:36:49
...余下省略...
'''
来源:https://www.cnblogs.com/gdjlc/p/11432526.html
猜你喜欢
- Msg 102, Level 15, State 1, Line 3 Incorrect syntax near '+'.
- 写在前面:本节主要讲述一些基本的,常用的SQL语句,而非数据库方面的基本知识。数据库方面的知识留待以后再讲,现在就我们平常常用的一些SQL语
- 如何在ADO中使用SQL函数?代码见下:<%Set conn1 = Server.CreateObjec
- 从2004年开始,我开始进入雅虎的异常表现小组。我们是一个很小的队伍,专门针对雅虎的产品进行质量检测和改进,我作为一个后端工程师,现在却开始
- 印象中最早看老外个人网站就挺纳闷,怎么人家都没有www,这样也可以?经过不断尝试,我发现确实不录入www要快捷的多,但不清楚怎么能做到。几年
- 今天帮一同学导入一个excel数据,我把excel保存为txt格式,然后建表导入,失败!分隔符格式不匹配,无法导入!!!!怎么看两边都是\t
- (一) 常用的CSS命名规则:头:header内容:content/container尾:footer导航:nav侧栏:sidebar栏目:
- 哎,以前写博文的时候没注意,有些图片用QQ来截取,获得的图片文件名都是类似于QQ截图20120926174732-300×15.png的形式
- 桥接模式(Bridge Pattern)是什么桥接模式是一种结构型模式,它将抽象部分与实现部分分离开来,使它们可以独立地变化。在桥接模式中,
- 准确地讲,Python没有专门处理字节的数据类型。但由于str既是字符串,又可以表示字节,所以,字节数组=str。而在C语言中,我们可以很方
- 简介使用 Python 进行数据分析时,比较常用的库有 Numpy、Pandas、Matplotlib,本篇文章就来说一下 Numpy 的使
- 出自: 编程中国 http://www.bc-cn.net作者: 天涯听雨 &nbs
- 使用Python的第三方库pyautogui,PyAutoGUI是一个纯Python的GUI自动化工具,其目的是可以用程序自动控制鼠标和键盘
- MySQL 一级防范检查列表以下是加固你的 Mysql 服务器安全所要做的工作的重要参考:Securing MySQL: step-by-s
- 尽管 JavaScript 历史上使用冗长而令人生厌的代码块来标的特定浏览器的时期已经结束了,但是偶尔使用一些简单的代码块和对象检测来确保一
- 1 基本概念1.1 命名空间 (namespace)命名空间是变量名到对象的映射(name -> obj)。目前大多数的命名空间以类似
- 惭愧啊,今天写了个查找子串的Python程序被BS了…如果让你写一个程序检查字符串s2中是不是包含有s1。也许你会很直观的写下下面的代码:#
- 目录文件读写读文件写文件StringIO和BytesIOStringIOBytesIO操作文件和目录环境变量操作文件和目录文件读写读文件tr
- 升级了浏览器到IE9,今天进入公司网站后台突然发现有些页面进不去了,F12调试显示有JS错误:DOM Exception: INVALID_
- 有时候让了解放双手,让电脑来帮我们自动发一些我们想要发的消息,挺省力的,比如说白天写好了演讲稿,晚上要在群里进行文字演讲,那么我们就可以用脚