Python运维之定时任务模块APScheduler

前言：本博客仅作记录学习使用，部分图片出自网络，如有侵犯您的权益，请联系删除

目录

定时任务模块APScheduler

一、安装及基本概念

1.1、APScheduler的安装

1.2、涉及概念

1.3、APScheduler的工作流程​编辑

二、配置调度器

三、启动调度器

四、调度事件监听

---

定时任务模块APScheduler

APScheduler提供了基于日期、固定时间间隔以及crontab类型的任务，我们可以在主程序的运行过程中快速增加新作业或删除旧作业。如果把作业存储在数据库中，那么作业的状态会被保存，当调度器重启时，不必重新添加作业，作业会恢复原状态继续执行。

一、安装及基本概念

1.1、APScheduler的安装

 pip install apscheduler

1.2、涉及概念

• 触发器（triggers）：触发器包含调度逻辑，描述一个任务何时被触发，有按日期、按时间间隔、按cronjob描述式三种触发方式。每个作业都有自己的触发器，除了初始配置之外，触发器是完全无状态的。
• 作业存储器（job stores）：指定了作业被存放的位置，默认的作业存储器是内存，也可以将作业保存在各种数据库中。当作业被存放在数据库中时，它会被序列化；当重新被加载时，会反序列化。作业存储器充当保存、加载、更新和查找作业的中间商。在调度器之间不能共享作业存储
• 执行器（executors）：执行器是将指定的作业（调用函数）提交到线程池或进程池中运行，当任务完成时，执行器通知调度器触发相应的事件。
• 调度器（schedulers）：任务调度器，控制器角色，通过它配置作业存储器、执行器和触发器、添加、修改和删除任务。调度器协调触发器、作业存储器、执行器的运行，通常只有一个调度程序运行在应用程序中，开发人员不需要直接处理作业存储器、执行器或触发器。配置作业存储器和执行器是通过调度器来完成的

1.3、APScheduler的工作流程

一个简单的间隔任务实例：

 import osfrom datetime import datetimefrom apscheduler.schedulers.blocking import BlockingScheduler​# 打印当前的时间def tick():print('Tick! The time is: %s' % datetime.now())​if __name__ == '__main__':scheduler = BlockingScheduler()# 添加一个作业rick，触发器为interval，每隔3秒执行一次scheduler.add_job(tick, 'interval', seconds=3)print('Press Ctrl+{0} to exit'.format('Break' if os.name == 'nt' else 'C'))try:scheduler.start()except (KeyboardInterrupt, SystemExit):pass

另外的触发器为date，cron。date按特定时间点触发，cron则按固定的时间间隔触发。

上述代码稍作修改可变为cron类的定时任务：

 import osfrom datetime import datetimefrom apscheduler.schedulers.blocking import BlockingScheduler​def tick():print('Tick! The time is: %s' % datetime.now())​if __name__ == '__main__':scheduler = BlockingScheduler()scheduler.add_job(tick, 'cron', hour=19,minute=23)print('Press Ctrl+{0} to exit'.format('Break' if os.name == 'nt' else 'C'))try:scheduler.start()except (KeyboardInterrupt, SystemExit):pass

定时cron任务也非常简单，直接给触发器trigger传入'cron'即可。hour=19，minute23，表示每天的19时23分执行任务

 hour=19,minute=23hour='19',minute='23'minute='*/3'  # 表示每3分钟执行一次hour='19-21',minute='23'  # 表示19:23、20:23、21:23各执行一次任务

二、配置调度器

调度器的主循环其实就是反复检查是否有到期需要执行的任务，具体分两步进行

• 询问自己的每一个作业存储器，有没有到期需要执行的任务。如果有则计算这些作业中每个作业需要 运行的时间点；如果时间点有多个，就做coalesce检查。

• 提交给执行器按时间点运行

各调度器的适用场景：

• BlockingSchduler：适用于调度程序，是进程中唯一运行的进程，调用start函数会阻塞当前线程，不能立即返回
• BackgroundScheduler：适用于调度程序，在应用程序的后台运行，调用start后主线程不会阻塞。
• AsyncIOScheduler：适用于使用了asyncio模块的应用程序
• GeventScheduler：适用于使用了gevent模块的应用程序
• TwistedScheduler：适用于构建Twisted的应用程序
• QtSchuduler：适用于构建Qt的应用程序。

作业存储器的选择：一是内存（ 默认），而是数据库。

执行器的选择：默认的ThreadPoolExecutor足够OK，如果作业负载涉及CPU密集型操作，那么考虑使用ProcessPoolExecutor，甚至同时使用，将其作为二级执行器。

APScheduler可以使用字典，关键字参数传递配置调度器。首先实例化调度程序添加作业，然后配置调度器，获得最大的灵活性。

如果调度程序在应用程序的后台运行，则选择BackgroundScheduler，并使用默认的jobstore和executor

 from apscheduler.schedulers.blocking import BlockingSchedulerscheduler = BlockingScheduler()

如果想配置更多的信息，就可设置两个执行器、两个作业存储器、调整新作业的默认值，并设置不同的时区。配置详情：

• 配置名为mongo的MongoDBjobStore作业存储器
• 配置名为default的SQLAlchemyJobStore（使用SQLite）
• 配置名为default的ThreadPoolExecutor，最大进程数为5
• UTC作为调度器的时区
• coalesce默认情况下关闭
• 作业的默认最大运行实例限制为3

方法一：

 from pytz import utc​from apscheduler.schedulers.background import BlockingSchedulerfrom apscheduler.jobstores.mongodb import MongoDBJobStorefrom apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStorefrom apscheduler.executors.pool import ThreadPoolExecutor,ProcessPoolExecutor​jobstores = {'mongo':MongoDBJobStore(),'default':SQLAlchemyJobStore(url='sqlite:///jobs.sqlite')}executors = {'default':ThreadPoolExecutor(20),'processpool':ProcessPoolExecutor(5)}job_defaults = {'coalesce':False,'max_instances':3}scheduler = BlockingScheduler(jobstores=jobstores,executors=executors,job_defaults=job_defaults,timezone=utc)

方法二：

 from apscheduler.schedulers.background import BlockingSchedulerscheduler = BlockingScheduler({'apscheduler.jobstores.mongo':{'type':'mongodb'},'apscheduler.jobstores.default':{'type':'sqlalchemy','url':'sqlite:///jobs.sqlite'},'apscheduler.executors.default':{'class':'apscheduler.executors.pool:ThreadPoolExecutor','max_workers':'5'},'apscheduler.job_defaults.coalesce':'fasle','apscheduler.job_defaults.max_instances':'3','apscheduler.timezone':'UTC',})

方法三：

 from pytz import utcfrom apscheduler.schedulers.background import BlockingSchedulerfrom apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStorefrom apscheduler.executors.pool import ThreadPoolExecutor,ProcessPoolExecutor​jobstores = {'mongo':{'type':'mongodb'},'default':SQLAlchemyJobStore(url='sqlite:///jobs.sqlite')}executors = {'default':{'type':'threadpool','max_workers':20},'processpool':ProcessPoolExecutor(max_workers=5)}job_defaults = {'coalesce':False,'max_instances':3}scheduler = BlockingScheduler()scheduler.configure(jobstores=jobstores,executors=executors,job_defaults=job_defaults,timezone=utc)

三、启动调度器

启动调度器前需要先添加作业，有两种方法可以向调度器添加作业：一是通过接口add_job()；二是通过使用函数装饰器，其中add_job()返回一个apscheduler.job.Job类的实例，用于后续修改或删除作业。

可以随时在调度器上调度作业。如果在添加作业时，调度器还没有启动，那么任务不会运行，并且它的第一次运行时间在调度器启动时计算。

调用调度器的start()方法启动调度器，下面用不同的作业存储器来举例：

 from apscheduler.schedulers.blocking import BlockingSchedulerimport datetimefrom apscheduler.jobstores.memory import MemoryJobStorefrom apscheduler.executors.pool import ThreadPoolExecutor,ProcessPoolExecutorf​def my_job(id='my_job'):print(id,'-->',datetime.datetime.now())​jobstores = {'default':MemoryJobStore()}executors = {'default':ThreadPoolExecutor(20),'processpool':ProcessPoolExecutor(10)}job_defaults = {'coalesce':False,'max_instance':3}scheduler =BlockingScheduler(jobstores=jobstores,executors=executors,job_defaults=job_defaults)scheduler.add_job(my_job,args=['job_interval',],id='job_interval',trigger='interval',seconds=5,replace_existing=True)scheduler.add_job(my_job,args=['job_cron',],id='job_cron',trigger='cron',month='4-8,5-6',hour='7-11',second='*/10',end_date='2024-06-06')scheduler.add_job(my_job,args=['job_once_now',],id='job_once_now')scheduler.add_job(my_job,args=['job_date_once',],id='job_date_once',trigger='date',run_date='2024-01-01 00:00:00')try:scheduler.start()except SystemExit:print('exit')exit()

方法二：使用数据库作为作业存储器（修改第5行和11行）

 from apscheduler.schedulers.blocking import BlockingSchedulerimport datetimefrom apscheduler.jobstores.memory import MemoryJobStorefrom apscheduler.executors.pool import ThreadPoolExecutor,ProcessPoolExecutorfrom apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore​def my_job(id='my_job'):print(id,'-->',datetime.datetime.now())​jobstores = {'default':SQLAlchemyJobStore(url='sqlite:///jobs.sqlite')}executors = {'default':ThreadPoolExecutor(20),'processpool':ProcessPoolExecutor(10)}job_defaults = {'coalesce':False,'max_instance':3}scheduler =BlockingScheduler(jobstores=jobstores,executors=executors,job_defaults=job_defaults)scheduler.add_job(my_job,args=['job_interval',],id='job_interval',trigger='interval',seconds=5,replace_existing=True)scheduler.add_job(my_job,args=['job_cron',],id='job_cron',trigger='cron',month='4-8,5-6',hour='7-11',second='*/10',end_date='2024-06-06')scheduler.add_job(my_job,args=['job_once_now',],id='job_once_now')scheduler.add_job(my_job,args=['job_date_once',],id='job_date_once',trigger='date',run_date='2024-01-01 00:00:00')try:scheduler.start()except SystemExit:print('exit')exit()

运行过之后，如果不注释添加作业的代码，则作业会重新添加到数据库中，这样就有了两个作业，为了避免这样的情况：设置（replace_existing=True）

 scheduler.add_job(my_job,args=['job_interval',],id='job_interval',trigger='interval',seconds=5,replace_existing=True)

如果想运行错过运行的作业，则使用misfire_grace_time：

 scheduler.add_job(my_job,args=['job_cron',],id='job_cron',trigger='cron',month='4-8,5-6',hour='7-11',second='*/10',coalesce=True,misfire_grace_time=30,replace_existing=True,end_date='2024-06-06')

其他操作如下：

 scheduler.remove_job(job_id,jobstore=None)      # 删除作业scheduler.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)      # 输出作业信息

四、调度事件监听

日志记录和事件监听：

 from apscheduler.schedulers.blocking import BlockingSchedulerfrom apscheduler.events import EVENT_JOB_EXECUTED,EVENT_JOB_ERRORimport datetimeimport logging​# 配置日志记录信息 logging.basicConfig(level=logging.INFO,format='%(asctime)s %(filename)s[line:%(lineno)d %(levelname)s %(message)s',datefmt='%Y-%m-%d %H:%M:%S',filename='log1.txt',filemode='a')​def aps_test(x):print(datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S'),x)​def date_test(x):print(datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S'),x)print(1/0)​def my_listener(event):if event.exception:print('任务出错了!!!!')else:print('任务照常运行...')​scheduler = BlockingScheduler()scheduler.add_job(func=date_test,args=('一次性任务，会出错',),next_run_time=datetime.datetime.now() + datetime.timedelta(seconds=15),id='date_task')scheduler.add_job(func=aps_test,args=('循环任务',),trigger='interval',seconds=3,id='interval_task')scheduler.add_listener(my_listener,EVENT_JOB_EXECUTED | EVENT_JOB_ERROR)scheduler._logger = logging​scheduler.start()

---

致谢

在此，我要对所有为知识共享做出贡献的个人和机构表示最深切的感谢。同时也感谢每一位花时间阅读这篇文章的读者，如果文章中有任何错误，欢迎留言批评指正。

学习永无止境，让我们共同进步