一文详解Python定时任务触发
APScheduler
APScheduler 四个组件分别为:
调度器(scheduler)、触发器(trigger),作业存储(job store),执行器(executor)
安装命令:
pip install setuptools pip install --ignore-installed apscheduler
1.新建调度器schedulers
BlockingScheduler : 调度器在当前进程的主线程中运行,也就是会阻塞当前线程
BackgroundScheduler : 调度器在后台线程中运行,不会阻塞当前线程
import datetime as dt from apscheduler.schedulers.blocking import BlockingScheduler scheduler = BlockingScheduler()
2.添加调度任务trigger
① date 触发器:(指定时间点触发),参数如下:
- run_date(datetime或str):任务运行的日期或时间
- timezone(datetime.tzinfo或str):指定时区
# 例1:在 2020-9-24 时刻运行一次 func 方法 scheduler.add_job(func, 'date', run_date = dt.date(2020, 9, 24)) # 例2: 在 2020-9-24 15:10:00 时刻运行一次 func 方法 scheduler.add_job(func, 'date', run_date = dt.datetime(2020, 9, 24, 15, 10, 0)) # 例3: 在 2020-9-24 15:11:00 时刻运行一次 func 方法 scheduler.add_job(func, 'date', run_date = '2020-9-24 15:11:00')
② interval 触发器: (固定时间间隔触发),参数如下:
- weeks(int):间隔几周
- days(int):间隔几天
- hours(int):间隔几小时
- minutes(int):间隔几分钟
- seconds(int):间隔几秒钟
- start_date(datetime或str):开始时间
- end_date(datetime或str):结束时间
- timezone(datetime.tzinfo或str):时区
# 例1:每隔两分钟执行一次 func 方法 scheduler.add_job(func, 'interval', minutes = 2) # 例2:在 2020-9-24 15:15:00 ~ 2020-9-24 15:20:00 之间, 每隔两分钟执行一次 func 方法 scheduler.add_job(func, 'interval', minutes = 2, start_date = '2020-9-24 15:15:00' , end_date = '2020-9-24 15:20:00')
③ cron 触发器:(在指定时间周期性地触发),参数如下:
- year(int 或 str):年
- month(int 或 str):月
- day(int 或 str):日
- week(int 或 str):周(1-53)
- day_of_week(int 或 str):星期几(0-6)
- hour(int 或 str):时
- minute(int 或 str):分
- second(int 或 str):秒
- start_date(datetime或str):最早开始时间(包含)
- end_date(datetime或str):最晚结束时间(包含)
- timezone(datetime.tzinfo或str):指定时区
字符 :
1. * 每一(每一分)
2. ? 表示不关心,任意
3. - 范围 (小时:1-12,1到12点运行)
4. , 标示多个值 (小时 1,2,3 1点2点3点运行)
5. / 递增触发(0/15,从0开始每15秒运行一次)
6. L 最后(日L,当月最后一天,周L周六)
7. W 指定日期最近的工作日(周一到周五)
8. # 序号(表示每月的第几个周几)
# 例:在每年 1-3、7-9 月份中的每个星期一、二中的 00:00, 01:00, 02:00 和 03:00 执行 func 任务 scheduler.add_job(func, 'cron', month = '1-3,7-9',day='0, tue', hour='0-3')
3.运行调度任务
scheduler.start()
3.1 测试时间
def forecast_adjust(): now_temp = datetime.now() print('执行方案一', now_temp, '时间间隔: ', now_temp-t0) def for2(): now_temp = datetime.now() print('执行方案二', now_temp, '时间间隔: ', now_temp-t0) def fortime3(): now_temp = datetime.now() print('执行方案三', now_temp, '时间间隔: ', now_temp-t0) return '9999999999999' def a__(): b = scheduler.add_job(fortime3, 'cron', hour='15', minute = '18') c = scheduler.add_job(fortime3, 'cron', hour='15', minute = '30') d = scheduler.add_job(fortime3, 'cron', hour='15', minute = '45') print(b) print(c) print(c) return 'kkkqq' t0 = datetime.now() scheduler = BlockingScheduler() # 采用阻塞的方式 scheduler.add_job(func=forecast_adjust, trigger=CronTrigger(minute="*/1", second=20, timezone=tz_now), args=[]) scheduler.add_job(func=for2, trigger=CronTrigger(minute="*/5", second=10, timezone=tz_now), args=[]) k = a__() print(k) scheduler.start()
4.特点,其他操作
APScheduler 定点、定时:
四个组件分别为:触发器(trigger),作业存储器(job store),执行器(executor),调度器(scheduler)
(1)job stores:对调度任务的管理:
① 添加job:
# add_job():可以改变或者移除 job scheduler.add_job(func, 'interval', minutes = 2) # scheduled_job():只适用于应用运行期间不会改变的 job scheduler.scheduled_job(func, 'interval', minutes = 2)
②移除job:
# remove_job() :根据 job 的 id 来移除,所以要在 job 创建的时候指定一个 id scheduler.add_job(func, 'interval', minutes = 2, id = 'job_one') scheduler.remove_job(job_one) # job.remove() :对 job 执行 remove 方法 job = add_job(func, 'interval', minutes = 2, id = 'job_one') job.remvoe()
③ 暂停job:
apscheduler.job.Job.pause() apscheduler.schedulers.base.BaseScheduler.pause_job()
④ 恢复job:
apscheduler.job.Job.resume() apscheduler.schedulers.base.BaseScheduler.resume_job()
⑤ 修改job:
# modify_job() scheduler.modify_job('job_one', minutes = 5) # job.modify() job = scheduler.add_job(func, 'interval', minutes = 2) job.modify(minutes = 5)
⑥ 关闭job:
scheduler.shutdown() scheduler.shutdown(wait=false)
(2)executors:执行调度任务的模块,常用的 executor 有两种:
ProcessPoolExecutor ThreadPoolExecutor
到此这篇关于一文详解Python定时任务触发的文章就介绍到这了,更多相关Python定时任务内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家!
相关文章
PyTorch实现ResNet50、ResNet101和ResNet152示例
今天小编就为大家分享一篇PyTorch实现ResNet50、ResNet101和ResNet152示例,具有很好的参考价值,希望对大家有所帮助。一起跟随小编过来看看吧2020-01-01python3的一个天坑问题及解决方法:报错UnicodeDecodeError: ‘utf-8‘
在调试程序发现python3的一个天坑问题:报错UnicodeDecodeError: ‘utf-8‘ codec can‘t decode byte 0xa3 in position 59: invalid,特此曝光,为众位开发朋友提个醒2023-09-09Python matplotlib绘制散点图配置(万能模板案例)
这篇文章主要介绍了Python matplotlib绘制散点图配置(万能模板案例),散点图是指在 回归分析中,数据点在直角坐标系平面上的分布图,散点图表示因变量随自变量而变化的大致趋势,据此可以选择合适的函数对数据点进行拟合2022-07-07Python Django给admin添加Action的方法实例详解
这篇文章主要介绍了Django给admin添加Action的方法,本文通过实例代码给大家介绍的非常详细,具有一定的参考借鉴价值,需要的朋友可以参考下2019-04-04解决pycharm运行程序出现卡住scanning files to index索引的问题
今天小编就为大家分享一篇解决pycharm运行程序出现卡住scanning files to index索引的问题,具有很好的参考价值,希望对大家有所帮助。一起跟随小编过来看看吧2019-06-06
最新评论