文章详情

短信预约-IT技能 免费直播动态提醒

请输入下面的图形验证码

提交验证

短信预约提醒成功

Python - 定时任务框架【APScheduler】基本使用详解(一)

2023-09-06 19:55

关注

一. 前言

一个网页会有很多数据是不需要经常变动的,比如说首页,变动频率低而访问量大,我们可以把它静态化,这样就不需要每次有请求都要查询数据库再返回,可以减少服务器压力
我们可以使用Django的模板渲染功能完成页面渲染

二. APSchedule简介

APScheduler的全称是Advanced Python Scheduler。它是一个轻量级的 Python 定时任务调度框架。
APScheduler 支持三种调度任务:

三. 使用

1.安装APScheduler

pip install APScheduler

2. 使用APScheduler 步骤

APScheduler 使用起来还算是比较简单。运行一个调度任务只需要以下三部曲:

  1. 新建一个 schedulers (调度器)
  2. 添加一个调度任务(job stores)
  3. 运行调度任务

简单示例
在项目模块或者根目录的__init__.py文件中添加以下代码

from apscheduler.scheduler import Schedulersched = Scheduler()@sched.interval_schedule(seconds=2,misfire_grace_time=3600)def excute_task():    fun()  #执行任务函数sched.start()  #启动定时任务脚本

启动项目即可自动执行

python manage.py runserver 8000

3. 基础组件介绍

APScheduler 有四种组件,分别是:调度器(scheduler),作业存储(job store),触发器(trigger),执行器(executor)。

schedulers(调度器)

提供 7种调度器,能够满足我们各种场景的需要。例如:后台执行某个操作,异步执行操作等。经常使用的为BackgroundScheduler
调度器分别是:

triggers(触发器)

APScheduler 有三种内建的 trigger

参数说明
run_date (datetime 或 str)作业的运行日期或时间
timezone (datetime.tzinfo 或 str)指定时区

date 触发器使用示例如下:

from datetime import datetimefrom datetime import datefrom apscheduler.schedulers.background import BackgroundSchedulerdef job_func(text):    print(text)scheduler = BackgroundScheduler()# 在 2017-12-13 时刻运行一次 job_func 方法scheduler.add_job(job_func, 'date', run_date=date(2017, 12, 13), args=['text'])# 在 2017-12-13 14:00:00 时刻运行一次 job_func 方法scheduler.add_job(job_func, 'date', run_date=datetime(2017, 12, 13, 14, 0, 0), args=['text'])# 在 2017-12-13 14:00:01 时刻运行一次 job_func 方法scheduler.add_job(job_func, 'date', run_date='2017-12-13 14:00:01', args=['text'])scheduler.start()

interval 触发器

固定时间间隔触发。interval 间隔调度

import datetimefrom apscheduler.schedulers.background import BackgroundSchedulerdef job_func(text):    print(datetime.datetime.utcnow().strftime("%Y-%m-%d %H:%M:%S.%f")[:-3])scheduler = BackgroundScheduler()# 每隔两分钟执行一次 job_func 方法scheduler.add_job(job_func, 'interval', minutes=2)# 在 2017-12-13 14:00:01 ~ 2017-12-13 14:00:10 之间, 每隔两分钟执行一次 job_func 方法scheduler.add_job(job_func, 'interval', minutes=2, start_date='2017-12-13 14:00:01' , end_date='2017-12-13 14:00:10')scheduler.start()

cron 触发器

在特定时间周期性地触发,和Linux crontab格式兼容。它是功能最强大的触发器。

import datetimefrom apscheduler.schedulers.background import BackgroundSchedulerdef job_func(text):    print("当前时间:", datetime.datetime.utcnow().strftime("%Y-%m-%d %H:%M:%S.%f")[:-3])scheduler = BackgroundScheduler()# 在每年 1-3、7-9 月份中的每个星期一、二中的 00:00, 01:00, 02:00 和 03:00 执行 job_func 任务scheduler.add_job(job_func, 'cron', month='1-3,7-9',day='0, tue', hour='0-3')scheduler.start()

四. 作业存储(job store)

该组件是对调度任务的管理

添加 job

有两种添加方法,其中一种上述代码用到的 add_job(), 另一种则是scheduled_job()修饰器来修饰函数。

这个两种办法的区别是:
第一种方法返回一个 apscheduler.job.Job 的实例,可以用来改变或者移除 job。
第二种方法只适用于应用运行期间不会改变的 job。

第二种添加任务方式的例子:

import datetimefrom apscheduler.schedulers.background import BackgroundScheduler@scheduler.scheduled_job(job_func, 'interval', minutes=2)def job_func(text):    print(datetime.datetime.utcnow().strftime("%Y-%m-%d %H:%M:%S.%f")[:-3])scheduler = BackgroundScheduler()scheduler.start()

移除 job

移除 job 也有两种方法:remove_job() 和 job.remove()。
remove_job() 是根据 job 的 id 来移除,所以要在 job 创建的时候指定一个 id。
job.remove() 则是对 job 执行 remove 方法即可

scheduler.add_job(job_func, 'interval', minutes=2, id='job_one')scheduler.remove_job(job_one)job = add_job(job_func, 'interval', minutes=2, id='job_one')job.remvoe()

获取 job 列表

通过 scheduler.get_jobs() 方法能够获取当前调度器中的所有 job 的列表

修改 job

如果你因计划改变要对 job 进行修改,可以使用Job.modify() 或者 modify_job()方法来修改 job 的属性。但是值得注意的是,job 的 id 是无法被修改的。

scheduler.add_job(job_func, 'interval', minutes=2, id='job_one')scheduler.start()# 将触发时间间隔修改成 5分钟scheduler.modify_job('job_one', minutes=5)job = scheduler.add_job(job_func, 'interval', minutes=2)# 将触发时间间隔修改成 5分钟job.modify(minutes=5)

关闭 job

默认情况下调度器会等待所有正在运行的作业完成后,关闭所有的调度器和作业存储。如果你不想等待,可以将 wait 选项设置为 False。

scheduler.shutdown()scheduler.shutdown(wait=false)

五.执行器(executor)

执行器顾名思义是执行调度任务的模块。最常用的 executor 有两种:ProcessPoolExecutorThreadPoolExecutor

下面是显式设置 job store(使用mongo存储)和 executor 的代码的示例。

代码示例

from pymongo import MongoClientfrom apscheduler.schedulers.blocking import BlockingSchedulerfrom apscheduler.jobstores.mongodb import MongoDBJobStorefrom apscheduler.jobstores.memory import MemoryJobStorefrom apscheduler.executors.pool import ThreadPoolExecutor, ProcessPoolExecutor  def my_job():    print 'hello world'host = '127.0.0.1'port = 27017client = MongoClient(host, port) jobstores = {    'mongo': MongoDBJobStore(collection='job', database='test', client=client),    'default': MemoryJobStore()}executors = {    'default': ThreadPoolExecutor(10),    'processpool': ProcessPoolExecutor(3)}job_defaults = {    'coalesce': False,    'max_instances': 3}scheduler = BlockingScheduler(jobstores=jobstores, executors=executors, job_defaults=job_defaults)scheduler.add_job(my_job, 'interval', seconds=5) try:    scheduler.start()except SystemExit:    client.close()

以上就是python- 定时任务框架【APScheduler】基本使用介绍,希望对你有所帮助!

来源地址:https://blog.csdn.net/qq_43030934/article/details/131421631

阅读原文内容投诉

免责声明:

① 本站未注明“稿件来源”的信息均来自网络整理。其文字、图片和音视频稿件的所属权归原作者所有。本站收集整理出于非商业性的教育和科研之目的,并不意味着本站赞同其观点或证实其内容的真实性。仅作为临时的测试数据,供内部测试之用。本站并未授权任何人以任何方式主动获取本站任何信息。

② 本站未注明“稿件来源”的临时测试数据将在测试完成后最终做删除处理。有问题或投稿请发送至: 邮箱/279061341@qq.com QQ/279061341

软考中级精品资料免费领

  • 历年真题答案解析
  • 备考技巧名师总结
  • 高频考点精准押题
  • 2024年上半年信息系统项目管理师第二批次真题及答案解析(完整版)

    难度     807人已做
    查看
  • 【考后总结】2024年5月26日信息系统项目管理师第2批次考情分析

    难度     351人已做
    查看
  • 【考后总结】2024年5月25日信息系统项目管理师第1批次考情分析

    难度     314人已做
    查看
  • 2024年上半年软考高项第一、二批次真题考点汇总(完整版)

    难度     433人已做
    查看
  • 2024年上半年系统架构设计师考试综合知识真题

    难度     221人已做
    查看

相关文章

发现更多好内容

猜你喜欢

AI推送时光机
位置:首页-资讯-后端开发
咦!没有更多了?去看看其它编程学习网 内容吧
首页课程
资料下载
问答资讯