Python任务调度模块APScheduler(内含定点报时案例)
Posted 小呆瓜耶
tags:
篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了Python任务调度模块APScheduler(内含定点报时案例)相关的知识,希望对你有一定的参考价值。
目录
- 总结
APScheduler介绍
提供了基于日期、固定时间间隔以及 crontab类型的任务。还可以在程序运行过程中动态的新增任务和删除任务。在任务运行过程中,还可以把任务存储起来,下次启动运行依然保留之前的状态。另外最重要的一个特点是,因为他是基于 Python语言的库,所以是可以跨平台的,一段代码,处处运行!
四个基本对象:
- 触发器(triggers):触发器就是根据你指定的触发方式,比如是按照时间间隔,还是按照 crontab触发,触发条件是什么等。每个任务都有自己的触发器。
- date:指定日期执行
- interval:可以指定具体间隔多少时间执行一次
- cron:可以指定执行的日期策略(最强大)
- 任务存储器(job stores):任务存储器是可以存储任务的地方,默认情况下任务保存在内存,也可将任务保存在各种数据库中。任务存储进去后,会进行序列化,然后也可以反序列化提取出来,继续执行。
- 执行器(executors):执行器的目的是安排任务到线程池或者进程池中运行的。
- 调度器(schedulers):任务调度器是属于整个调度的总指挥官。他会合理安排作业存储器、执行器、触发器进行工作,并进行添加和删除任务等。调度器通常是只有一个的。开发人员很少直接操作触发器、存储器、执行器等。因为这些都由调度器自动来实现了。
任务存储器
任务存储器的选择有两种。一是内存,也是默认的配置。二是数据库。使用内存的方式是简单高效,但是不好的是,一旦程序出现问题,重新运行的话,会把之前已经执行了的任务重新执行一遍。数据库则可以在程序崩溃后,重新运行可以从之前中断的地方恢复正常运行。有以下几种选择:
- MemoryJobStore :没有序列化,任务存储在内存中,增删改查都是在内存中完成。
- SQLAlchemyJobStore :使用 SQLAlchemy 这个 ORM 框架作为存储方式。
- MongoDBJobStore :使用 mongodb 作为存储器。
- RedisJobStore :使用 redis 作为存储器。
执行器
执行器的选择取决于应用场景。通常默认的 ThreadPoolExecutor已经在大部分情况下是可以满足我们需求的。如果我们的任务涉及到一些 CPU密集计算的操作。那么应该考虑 ProcessPoolExecutor。然后针对每种程序, apscheduler也设置了不同的 executor:
- ThreadPoolExecutor :线程池执行器。
- ProcessPoolExecutor :进程池执行器。
- GeventExecutor : Gevent 程序执行器。
- TornadoExecutor : Tornado 程序执行器。
- TwistedExecutor : Twisted 程序执行器。
- AsyncIOExecutor : asyncio 程序执行器。
调度器
- BlockingScheduler :适用于调度程序是进程中唯一运行的进程,调用 start 函数会阻塞当前线程,不能立即返回。
- BackgroundScheduler :适用于调度程序在应用程序的后台运行,调用 start 后主线程不会阻塞。
- Asyncioscheduler :适用于使用了 asyncio 模块的应用程序。
- GeventScheduler :适用于使用 gevent 模块的应用程序。
- TwistedScheduler :适用于构建 Twisted 的应用程序。
- QtScheduler :适用于构建 Qt 的应用程序。
定时任务调度配置
这里我们用一个例子来说明。比如我想这样配置
- 执行器:配置 default 执行器为 ThreadPoolExecutor ,并且设置最多的线程数是30个。
- 存储器:配置 default 的任务存储器为 SQLAlchemyJobStore (使用 SQLite ) 。
- 任务配置:设置 coalesce 为 False :设置这个目的是,比如由于某个原因导致某个任务积攒了很多次没有执行(比如有一个任务是1分钟跑一次,但是系统原因断了3分钟),如果 coalesce = True ,那么下次恢复运行的时候,会只执行一次,而如果设置 coalesce = False ,那么就不会合并,会3次全部执行。max_instances = 3 :同一个任务同一时间最多只能有3个实例在运行。比如一个耗时10分钟的job,被指定每分钟运行1次,如果我 max_instance 值3,那么在第 4 ~ 7 分钟上,新的运行实例不会被执行,因为已经有3个实例在跑了。
BackgroundScheduler & BlockingScheduler 的区别
BackgroundScheduler
后台调度,不会阻塞主程序执行
使用场景:适用于不等待任务调度执行结束,继续往下执行主程序。
# 导入后台调度模块
from apscheduler.schedulers.background import BackgroundScheduler
# 导入时间模块
import time
# job1任务
def job1():
print(我是job1,我每3秒执行一次)
# 主程序
def main():
while (True):
print(我是主程序,我每1秒执行一次)
# 休息一秒
time.sleep(1)
# 入口函数
if __name__ == __main__:
# 实例一个后台调度
scheduler = BackgroundScheduler(timezone=MST)
# 后台调度添加一个任务,每3秒执行一次
scheduler.add_job(job1, interval, id=3_second_job, seconds=3)
# 开始执行调度
scheduler.start()
# 主程序
main()
BlockingScheduler
前台调度(阻塞调度),会阻塞主程序的执行
使用场景:适用于必须等待任务调度执行结束,才往下执行主程序场景。
# 导入前台调度模块
from apscheduler.schedulers.blocking import BlockingScheduler
# 导入时间模块
import time
# job1任务
def job1():
print(我是job1,我每3秒执行一次)
# 主程序
def main():
while (True):
print(我是主程序,我每1秒执行一次)
# 休息一秒
time.sleep(1)
# 入口函数
if __name__ == __main__:
# 实例一个前台调度
scheduler = BlockingScheduler(timezone=MST)
# 前台调度添加一个任务,每3秒执行一次
scheduler.add_job(job1, interval, id=3_second_job, seconds=3)
# 开始执行调度
scheduler.start()
# 主程序(不会执行到这里)
main()
APScheduler具体使用
date触发器
# 导入系统操作模块
import os
# 导入时间模块
from datetime import datetime
# 导入前台调度模块
from apscheduler.schedulers.blocking import BlockingScheduler
def tell_time() -> None:
"""
报时
:return: None
"""
dt = datetime.now()
text = f现在时间是:dt
print(text)
# windows说话命令行Speak中为讲话内容,有单引号
md = fPowerShell -Command Add-Type -AssemblyName System.Speech; $speak=New-Object System.Speech.Synthesis.SpeechSynthesizer; $speak.rate=4; $speak.Speak(text);
os.system(md)
# 入口函数
if __name__ == __main__:
# 实例一个前台调度
scheduler = BlockingScheduler(timezone=MST)
# 指定2022年8月17日报时
scheduler.add_job(tell_time, date, rund_date=2022-8-17)
# 指定2022年8月17日报时
scheduler.add_job(tell_time, date, run_date=datetime(2022, 8, 17))
# 立即执行
scheduler.add_job(tell_time, date)
# 开始执行调度
scheduler.start()
intervalc触发器
# 导入系统操作模块
import os
# 导入时间模块
from datetime import datetime
# 导入前台调度模块
from apscheduler.schedulers.blocking import BlockingScheduler
def tell_time() -> None:
"""
报时
:return: None
"""
dt = datetime.now()
text = f现在时间是:dt
print(text)
# windows说话命令行Speak中为讲话内容,有单引号
md = fPowerShell -Command Add-Type -AssemblyName System.Speech; $speak=New-Object System.Speech.Synthesis.SpeechSynthesizer; $speak.rate=4; $speak.Speak(text);
os.system(md)
# 入口函数
if __name__ == __main__:
# 实例一个前台调度
scheduler = BlockingScheduler(timezone=MST)
# 间隔8秒报一次时,时间间隔过短如果上一个任务在执行中,会报下面一段错误(不会停止执行任务,等到下一次间隔时间再执行),意思是达到最大运行实例数
# Execution of job "tell_time (trigger: interval[0:00:06], next run at: 2022-08-16 06:58:02 MST)" skipped: maximum number of running instances reached (1)
scheduler.add_job(tell_time, interval, seconds=8)
# 间隔8分钟报时
scheduler.add_job(tell_time, interval, minutes=8)
# 间隔8小时报时
scheduler.add_job(tell_time, interval, hours=8)
# 间隔15天21小时14分钟2秒报时
scheduler.add_job(tell_time, interval, days=16, hours=22, minutes=14, seconds=2)
# 开始执行调度
scheduler.start()
interval 触发器参数
表达式 | 描述 |
weeks | 周 |
days | 1-31日 |
hour | 0 - 23 小时 |
minute | 0 - 59 分钟 |
second | 0 - 59 秒 |
start_date | 起始时间,datetime 类型或者字符串类型 |
end_date | 结束时间,datetime 类型或者字符串类型 |
jitter | 触发的时间误差 |
cron触发器
# 导入系统操作模块
import os
# 导入时间模块
from datetime import datetime
# 导入前台调度模块
from apscheduler.schedulers.blocking import BlockingScheduler
def tell_time() -> None:
"""
报时
:return: None
"""
dt = datetime.now()
text = f现在时间是:dt
print(text)
# windows说话命令行Speak中为讲话内容,有单引号
md = fPowerShell -Command Add-Type -AssemblyName System.Speech; $speak=New-Object System.Speech.Synthesis.SpeechSynthesizer; $speak.rate=4; $speak.Speak(text);
os.system(md)
# 入口函数
if __name__ == __main__:
# 实例一个前台调度
scheduler = BlockingScheduler(timezone=MST)
# 6点报时
scheduler.add_job(tell_time, cron, hour=21)
# 6分报时
scheduler.add_job(tell_time, cron, minute=48)
# 6秒报时
scheduler.add_job(tell_time, cron, second=6)
# 6点6分6秒报时
scheduler.add_job(tell_time, cron, hour=6, minute=6, second=6)
# 在2022年1月1日-2023年1月1日间的每周一到周五的6点6分执行
scheduler.add_job(tell_time, cron, day_of_week=mon-fri, hour=6, minute=6, start_date=2022-1-1,
end_date=2023-1-1)
# 开始执行调度
scheduler.start()
cron触发器参数
表达式 | 描述 |
year | 4位数字的年份 |
month | 1-12月份 |
day | 1-31日 |
week | 1-53周 |
day_of_week | 一个礼拜中的第几天( 0 - 6 或者 mon 、 tue 、 wed 、 thu 、 fri 、 sat 、 sun ) |
hour | 0 - 23 小时 |
minute | 0 - 59 分钟 |
second | 0 - 59 秒 |
start_date | 起始时间,datetime 类型或者字符串类型 |
end_date | 结束时间,datetime 类型或者字符串类型 |
timezone | 时区 |
jitter | 触发的时间误差 |
任务操作
- 添加任务:scheduler.add_job(job_obj,args,id,trigger,**trigger_kwargs)
- 删除任务:scheduler.remove_job(job_id,jobstore=None)
- 删除所有任务:remove_all_jobs(jobstore=None)
- 暂停任务:scheduler.pause_job(job_id,jobstore=None)
- 恢复任务:scheduler.resume_job(job_id,jobstore=None)
- 修改某个任务属性信息:scheduler.modify_job(job_id,jobstore=None,**changes)
- 修改单个作业的触发器并更新下次运行时间:scheduler.reschedule_job(job_id,jobstore=None,trigger=None,**trigger_args)
- 输出所有作业信息:scheduler.print_jobs(jobstore=None,out=sys.stdout)
- 获取所有job信息:scheduler.get_jobs()
# 导入系统操作模块
import os
# 导入时间模块
import sys
from datetime import datetime
# 导入后台调度模块
from apscheduler.schedulers.blocking import BlockingScheduler
def tell_time() -> None:
"""
报时
:return: None
"""
dt = datetime.now()
text = f现在时间是:dt
print(text)
# 获取所有job信息
jobs = scheduler.get_jobs()
print(get_jobs, jobs)
# 输出所有作业信息
scheduler.print_jobs(jobstore=None, out=sys.stdout)
# windows说话命令行Speak中为讲话内容,有单引号
md = fPowerShell -Command Add-Type -AssemblyName System.Speech; $speak=New-Object System.Speech.Synthesis.SpeechSynthesizer; $speak.rate=4; $speak.Speak(text);
os.system(md)
# 入口函数
if __name__ == __main__:
# 实例一个后台调度
scheduler = BlockingScheduler(timezone=MST)
# 添加任务
scheduler.add_job(tell_time, interval, id=1_seconds, seconds=1)
scheduler.add_job(tell_time, interval, id=60_seconds, seconds=60)
scheduler.add_job(tell_time, date, id=2022_date, run_date=datetime(2022, 8, 18))
scheduler.add_job(tell_time, cron, id=6_cron, hour=6, minute=6, second=6)
# 删除任务
scheduler.remove_job(1_seconds)
# 暂停任务
scheduler.pause_job(2022_date)
scheduler.pause_job(6_cron)
# 恢复任务
scheduler.resume_job(6_cron)
# 修改任务
scheduler.modify_job(job_id=2022_date, jobstore=None)
# 修改单个作业的触发器并更新下次运行时间
result = scheduler.reschedule_job(job_id=6_cron, trigger=interval, seconds=10)
# 开始执行调度
scheduler.start()
总结
APScheduler是一款非常强大的作业调度框架,支持跨平台,配合cron使用非常灵活,几乎支持日常所需的任务执行调度。APScheduler也可结合多线程,用于爬虫项目,或者定时消息通知,定时邮件等,后面我也会更新一篇以APScheduler结合爬虫的每天定时天气预报通知。
以上是关于Python任务调度模块APScheduler(内含定点报时案例)的主要内容,如果未能解决你的问题,请参考以下文章
Python任务调度模块APScheduler(内含定点报时案例)