代码收藏家技术教程 2024-08-09

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 os
 from datetime import datetime
 from 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 os
 from datetime import datetime
 from 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=23
 hour='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 BlockingScheduler
 scheduler = BlockingScheduler()

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

配置名为mongo的MongoDBjobStore作业存储器

配置名为default的SQLAlchemyJobStore（使用SQLite）

配置名为default的ThreadPoolExecutor，最大进程数为5

UTC作为调度器的时区

coalesce默认情况下关闭

作业的默认最大运行实例限制为3

方法一：

 from pytz import utc
 
 from apscheduler.schedulers.background import BlockingScheduler
 from apscheduler.jobstores.mongodb import MongoDBJobStore
 from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore
 from 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 BlockingScheduler
 scheduler = 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 utc
 from apscheduler.schedulers.background import BlockingScheduler
 from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore
 from 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 BlockingScheduler
 import datetime
 from apscheduler.jobstores.memory import MemoryJobStore
 from apscheduler.executors.pool import ThreadPoolExecutor,ProcessPoolExecutor
 f
 
 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 BlockingScheduler
 import datetime
 from apscheduler.jobstores.memory import MemoryJobStore
 from apscheduler.executors.pool import ThreadPoolExecutor,ProcessPoolExecutor
 from 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 BlockingScheduler
 from apscheduler.events import EVENT_JOB_EXECUTED,EVENT_JOB_ERROR
 import datetime
 import 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()