Python任务调度模块APScheduler(内含定点报时案例)

Posted 小呆瓜耶

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了Python任务调度模块APScheduler(内含定点报时案例)相关的知识,希望对你有一定的参考价值。



目录


APScheduler介绍

提供了基于日期、固定时间间隔以及 crontab类型的任务。还可以在程序运行过程中动态的新增任务和删除任务。在任务运行过程中,还可以把任务存储起来,下次启动运行依然保留之前的状态。另外最重要的一个特点是,因为他是基于 Python语言的库,所以是可以跨平台的,一段代码,处处运行!


四个基本对象:

  1. 触发器(triggers):触发器就是根据你指定的触发方式,比如是按照时间间隔,还是按照 crontab触发,触发条件是什么等。每个任务都有自己的触发器。
  • date:指定日期执行
  • interval:可以指定具体间隔多少时间执行一次
  • cron:可以指定执行的日期策略(最强大)
  1. 任务存储器(job stores):任务存储器是可以存储任务的地方,默认情况下任务保存在内存,也可将任务保存在各种数据库中。任务存储进去后,会进行序列化,然后也可以反序列化提取出来,继续执行。
  2. 执行器(executors):执行器的目的是安排任务到线程池或者进程池中运行的。
  3. 调度器(schedulers):任务调度器是属于整个调度的总指挥官。他会合理安排作业存储器、执行器、触发器进行工作,并进行添加和删除任务等。调度器通常是只有一个的。开发人员很少直接操作触发器、存储器、执行器等。因为这些都由调度器自动来实现了。

任务存储器

任务存储器的选择有两种。一是内存,也是默认的配置。二是数据库。使用内存的方式是简单高效,但是不好的是,一旦程序出现问题,重新运行的话,会把之前已经执行了的任务重新执行一遍。数据库则可以在程序崩溃后,重新运行可以从之前中断的地方恢复正常运行。有以下几种选择:

  1. MemoryJobStore :没有序列化,任务存储在内存中,增删改查都是在内存中完成。
  2. SQLAlchemyJobStore :使用 SQLAlchemy 这个 ORM 框架作为存储方式。
  3. MongoDBJobStore :使用 mongodb 作为存储器。
  4. RedisJobStore :使用 redis 作为存储器。

执行器

执行器的选择取决于应用场景。通常默认的 ThreadPoolExecutor已经在大部分情况下是可以满足我们需求的。如果我们的任务涉及到一些 CPU密集计算的操作。那么应该考虑 ProcessPoolExecutor。然后针对每种程序, apscheduler也设置了不同的 executor:

  1. ThreadPoolExecutor :线程池执行器。
  2. ProcessPoolExecutor :进程池执行器。
  3. GeventExecutor : Gevent 程序执行器。
  4. TornadoExecutor : Tornado 程序执行器。
  5. TwistedExecutor : Twisted 程序执行器。
  6. AsyncIOExecutor : asyncio 程序执行器。

调度器

  1. BlockingScheduler :适用于调度程序是进程中唯一运行的进程,调用 start 函数会阻塞当前线程,不能立即返回。
  2. BackgroundScheduler :适用于调度程序在应用程序的后台运行,调用 start 后主线程不会阻塞。
  3. Asyncioscheduler :适用于使用了 asyncio 模块的应用程序。
  4. GeventScheduler :适用于使用 gevent 模块的应用程序。
  5. TwistedScheduler :适用于构建 Twisted 的应用程序。
  6. QtScheduler :适用于构建 Qt 的应用程序。

定时任务调度配置

这里我们用一个例子来说明。比如我想这样配置

  1. 执行器:配置 default 执行器为 ThreadPoolExecutor ,并且设置最多的线程数是30个。
  2. 存储器:配置 default 的任务存储器为 SQLAlchemyJobStore (使用 SQLite ) 。
  3. 任务配置:设置 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(内含定点报时案例)

Python任务调度模块 – APScheduler

python中的apscheduler模块

django定时任务python调度框架APScheduler使用详解

Python3-apscheduler模块-定时调度

python中APScheduler的使用详解(python3经典编程案例)