上海做网站吧,企业免费建站,设计装修公司哪家好,站长统计 网站统计前言#xff1a;本博客仅作记录学习使用#xff0c;部分图片出自网络#xff0c;如有侵犯您的权益#xff0c;请联系删除 目录
定时任务模块APScheduler
一、安装及基本概念
1.1、APScheduler的安装
1.2、涉及概念
1.3、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, seconds3)print(Press Ctrl{0} to exit.format(Break if os.name nt else C))try:scheduler.start()except (KeyboardInterrupt, SystemExit):pass
另外的触发器为datecron。date按特定时间点触发cron则按固定的时间间隔触发。
上述代码稍作修改可变为cron类的定时任务 import osfrom datetime import datetimefrom apscheduler.schedulers.blocking import BlockingSchedulerdef tick():print(Tick! The time is: %s % datetime.now())if __name__ __main__:scheduler BlockingScheduler()scheduler.add_job(tick, cron, hour19,minute23)print(Press Ctrl{0} to exit.format(Break if os.name nt else C))try:scheduler.start()except (KeyboardInterrupt, SystemExit):pass
定时cron任务也非常简单直接给触发器trigger传入cron即可。hour19minute23表示每天的19时23分执行任务 hour19,minute23hour19,minute23minute*/3 # 表示每3分钟执行一次hour19-21,minute23 # 表示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最大进程数为5UTC作为调度器的时区coalesce默认情况下关闭作业的默认最大运行实例限制为3
方法一 from pytz import utcfrom apscheduler.schedulers.background import BlockingSchedulerfrom apscheduler.jobstores.mongodb import MongoDBJobStorefrom apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStorefrom apscheduler.executors.pool import ThreadPoolExecutor,ProcessPoolExecutorjobstores {mongo:MongoDBJobStore(),default:SQLAlchemyJobStore(urlsqlite:///jobs.sqlite)}executors {default:ThreadPoolExecutor(20),processpool:ProcessPoolExecutor(5)}job_defaults {coalesce:False,max_instances:3}scheduler BlockingScheduler(jobstoresjobstores,executorsexecutors,job_defaultsjob_defaults,timezoneutc)
方法二 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,ProcessPoolExecutorjobstores {mongo:{type:mongodb},default:SQLAlchemyJobStore(urlsqlite:///jobs.sqlite)}executors {default:{type:threadpool,max_workers:20},processpool:ProcessPoolExecutor(max_workers5)}job_defaults {coalesce:False,max_instances:3}scheduler BlockingScheduler()scheduler.configure(jobstoresjobstores,executorsexecutors,job_defaultsjob_defaults,timezoneutc)
三、启动调度器
启动调度器前需要先添加作业有两种方法可以向调度器添加作业一是通过接口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,ProcessPoolExecutorfdef my_job(idmy_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(jobstoresjobstores,executorsexecutors,job_defaultsjob_defaults)scheduler.add_job(my_job,args[job_interval,],idjob_interval,triggerinterval,seconds5,replace_existingTrue)scheduler.add_job(my_job,args[job_cron,],idjob_cron,triggercron,month4-8,5-6,hour7-11,second*/10,end_date2024-06-06)scheduler.add_job(my_job,args[job_once_now,],idjob_once_now)scheduler.add_job(my_job,args[job_date_once,],idjob_date_once,triggerdate,run_date2024-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 SQLAlchemyJobStoredef my_job(idmy_job):print(id,--,datetime.datetime.now())jobstores {default:SQLAlchemyJobStore(urlsqlite:///jobs.sqlite)}executors {default:ThreadPoolExecutor(20),processpool:ProcessPoolExecutor(10)}job_defaults {coalesce:False,max_instance:3}scheduler BlockingScheduler(jobstoresjobstores,executorsexecutors,job_defaultsjob_defaults)scheduler.add_job(my_job,args[job_interval,],idjob_interval,triggerinterval,seconds5,replace_existingTrue)scheduler.add_job(my_job,args[job_cron,],idjob_cron,triggercron,month4-8,5-6,hour7-11,second*/10,end_date2024-06-06)scheduler.add_job(my_job,args[job_once_now,],idjob_once_now)scheduler.add_job(my_job,args[job_date_once,],idjob_date_once,triggerdate,run_date2024-01-01 00:00:00)try:scheduler.start()except SystemExit:print(exit)exit()
运行过之后如果不注释添加作业的代码则作业会重新添加到数据库中这样就有了两个作业为了避免这样的情况设置replace_existingTrue scheduler.add_job(my_job,args[job_interval,],idjob_interval,triggerinterval,seconds5,replace_existingTrue)
如果想运行错过运行的作业则使用misfire_grace_time scheduler.add_job(my_job,args[job_cron,],idjob_cron,triggercron,month4-8,5-6,hour7-11,second*/10,coalesceTrue,misfire_grace_time30,replace_existingTrue,end_date2024-06-06)
其他操作如下 scheduler.remove_job(job_id,jobstoreNone) # 删除作业scheduler.remove_all_jobs(jobstoreNone) # 删除所有作业scheduler.pause_job(job_id,jobstoreNone) # 暂停作业scheduler.resume_job(job_id,jobstoreNone) # 恢复作业scheduler.modify_job(job_id,jobstoreNone,**changes) # 修改单个作业属性配置scheduler.reschedule_job(job_id,jobstoreNone,triggerNone,**trigger_args) # 修改单个作业的触发器并更新下次运行时间scheduler.print_jobs(jobstoreNone,outsys.stdout) # 输出作业信息
四、调度事件监听
日志记录和事件监听 from apscheduler.schedulers.blocking import BlockingSchedulerfrom apscheduler.events import EVENT_JOB_EXECUTED,EVENT_JOB_ERRORimport datetimeimport logging# 配置日志记录信息 logging.basicConfig(levellogging.INFO,format%(asctime)s %(filename)s[line:%(lineno)d %(levelname)s %(message)s,datefmt%Y-%m-%d %H:%M:%S,filenamelog1.txt,filemodea)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(funcdate_test,args(一次性任务会出错,),next_run_timedatetime.datetime.now() datetime.timedelta(seconds15),iddate_task)scheduler.add_job(funcaps_test,args(循环任务,),triggerinterval,seconds3,idinterval_task)scheduler.add_listener(my_listener,EVENT_JOB_EXECUTED | EVENT_JOB_ERROR)scheduler._logger loggingscheduler.start() 致谢
在此我要对所有为知识共享做出贡献的个人和机构表示最深切的感谢。同时也感谢每一位花时间阅读这篇文章的读者如果文章中有任何错误欢迎留言批评指正。
学习永无止境让我们共同进步
