Python使用APScheduler实现定时任务过程解析

前言

APScheduler是基于Quartz的一个Python定时任务框架。提供了基于日期、固定时间间隔以及crontab类型的任务,并且可以持久化任务。

在线文档:https://apscheduler.readthedocs.io/en/latest/userguide.html

一、安装APScheduler

pip install apscheduler

二、基本概念

APScheduler有四大组件:

1、触发器 triggers :
触发器包含调度逻辑。每个作业都有自己的触发器,用于确定下一个任务何时运行。除了初始配置之外,触发器是完全无状态的。

有三种内建的trigger:

  • (1)date: 特定的时间点触发
  • (2)interval: 固定时间间隔触发
  • (3)cron: 在特定时间周期性地触发

2、任务储存器 job stores:用于存放任务,把任务存放在内存(为默认MemoryJobStore)或数据库中。

3、执行器 executors: 执行器是将任务提交到线程池或进程池中运行,当任务完成时,执行器通知调度器触发相应的事件。

4、调度器 schedulers: 把上方三个组件作为参数,通过创建调度器实例来运行

根据开发需求选择相应的组件,下面是不同的调度器组件:

  • BlockingScheduler 阻塞式调度器:适用于只跑调度器的程序。
  • BackgroundScheduler 后台调度器:适用于非阻塞的情况,调度器会在后台独立运行。
  • AsyncIOScheduler AsyncIO调度器,适用于应用使用AsnycIO的情况。
  • GeventScheduler Gevent调度器,适用于应用通过Gevent的情况。
  • TornadoScheduler Tornado调度器,适用于构建Tornado应用。
  • TwistedScheduler Twisted调度器,适用于构建Twisted应用。
  • QtScheduler Qt调度器,适用于构建Qt应用。

三、使用步骤

1、新建一个调度器schedulers

2、添加调度任务

3、运行调度任务

四、使用实例

1、触发器date

特定的时间点触发,只执行一次。参数如下:

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

使用例子:

from datetime import datetime
from datetime import date
from apscheduler.schedulers.blocking import BlockingScheduler
def job(text):
 print(text)
scheduler = BlockingScheduler()
# 在 2019-8-30 运行一次 job 方法
scheduler.add_job(job, 'date', run_date=date(2019, 8, 30), args=['text1'])
# 在 2019-8-30 01:00:00 运行一次 job 方法
scheduler.add_job(job, 'date', run_date=datetime(2019, 8, 30, 1, 0, 0), args=['text2'])
# 在 2019-8-30 01:00:01 运行一次 job 方法
scheduler.add_job(job, 'date', run_date='2019-8-30 01:00:00', args=['text3'])
scheduler.start()

2、触发器interval

固定时间间隔触发。参数如下:

参数 说明
weeks (int) 间隔几周
days (int) 间隔几天
hours (int) 间隔几小时
minutes (int) 间隔几分钟
seconds (int) 间隔多少秒
start_date (datetime 或 str) 开始日期
end_date (datetime 或 str) 结束日期
timezone (datetime.tzinfo 或str)  

使用例子:

import time
from apscheduler.schedulers.blocking import BlockingScheduler
def job(text):
  t = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time()))
  print('{} --- {}'.format(text, t))
scheduler = BlockingScheduler()
# 每隔 1分钟 运行一次 job 方法
scheduler.add_job(job, 'interval', minutes=1, args=['job1'])
# 在 2019-08-29 22:15:00至2019-08-29 22:17:00期间,每隔1分30秒 运行一次 job 方法
scheduler.add_job(job, 'interval', minutes=1, seconds = 30, start_date='2019-08-29 22:15:00', end_date='2019-08-29 22:17:00', args=['job2'])
scheduler.start()
'''
运行结果:
job2 --- 2019-08-29 22:15:00
job1 --- 2019-08-29 22:15:46
job2 --- 2019-08-29 22:16:30
job1 --- 2019-08-29 22:16:46
job1 --- 2019-08-29 22:17:46
...余下省略...
'''

3、触发器cron

在特定时间周期性地触发。参数如下:

参数 说明
year (int 或 str) 年,4位数字
month (int 或 str) 月 (范围1-12)
day (int 或 str) 日 (范围1-31)
week (int 或 str) 周 (范围1-53)
day_of_week (int 或 str) 周内第几天或者星期几 (范围0-6 或者 mon,tue,wed,thu,fri,sat,sun)
hour (int 或 str) 时 (范围0-23)
minute (int 或 str) 分 (范围0-59)
second (int 或 str) 秒 (范围0-59)
start_date (datetime 或 str) 最早开始日期(包含)
end_date (datetime 或 str) 最晚结束时间(包含)
timezone (datetime.tzinfo 或str) 指定时区

这些参数支持算数表达式,取值格式有如下:

使用例子:

import time
from apscheduler.schedulers.blocking import BlockingScheduler
def job(text):
  t = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time()))
  print('{} --- {}'.format(text, t))
scheduler = BlockingScheduler()
# 在每天22点,每隔 1分钟 运行一次 job 方法
scheduler.add_job(job, 'cron', hour=22, minute='*/1', args=['job1'])
# 在每天22和23点的25分,运行一次 job 方法
scheduler.add_job(job, 'cron', hour='22-23', minute='25', args=['job2'])
scheduler.start()
'''
运行结果:
job1 --- 2019-08-29 22:25:00
job2 --- 2019-08-29 22:25:00
job1 --- 2019-08-29 22:26:00
job1 --- 2019-08-29 22:27:00
...余下省略...
'''

4、通过装饰器scheduled_job()添加方法

添加任务的方法有两种:

(1)通过调用add_job()---见上面1至3代码

(2)通过装饰器scheduled_job():

第一种方法是最常用的方法。第二种方法主要是方便地声明在应用程序运行时不会更改的任务。该 add_job()方法返回一个apscheduler.job.Job实例,可以使用该实例稍后修改或删除该任务。

import time
from apscheduler.schedulers.blocking import BlockingScheduler
scheduler = BlockingScheduler()
@scheduler.scheduled_job('interval', seconds=5)
def job1():
 t = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time()))
 print('job1 --- {}'.format(t))
@scheduler.scheduled_job('cron', second='*/7')
def job2():
 t = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time()))
 print('job2 --- {}'.format(t))
scheduler.start()
'''
运行结果:
job2 --- 2019-08-29 22:36:35
job1 --- 2019-08-29 22:36:37
job2 --- 2019-08-29 22:36:42
job1 --- 2019-08-29 22:36:42
job1 --- 2019-08-29 22:36:47
job2 --- 2019-08-29 22:36:49
...余下省略...
'''

以上就是本文的全部内容,希望对大家的学习有所帮助,也希望大家多多支持我们。

(0)

相关推荐

  • 详解python调度框架APScheduler使用

    最近在研究python调度框架APScheduler使用的路上,那么今天也算个学习笔记吧! # coding=utf-8 """ Demonstrates how to use the background scheduler to schedule a job that executes on 3 second intervals. """ from datetime import datetime import time import os

  • 详解Python下Flask-ApScheduler快速指南

    引言:Flask是Python社区非常流行的一个Web开发框架,本文将尝试将介绍APScheduler应用于Flask之中. 1. Flask介绍 Flask是Python社区大名鼎鼎的"microframework",基于简单的核心,使用extension来增加其他功能,其提供非常丰富易用的扩展包, 比如: 2.  Flask-APScheduler 社区提供了一个Flask-APScheduler的模块,方便大家直接在Flask模块中使用APScheduler. 关于安装的命令,仍

  • Python定时任务APScheduler的实例实例详解

    APScheduler 支持三种调度任务:固定时间间隔,固定时间点(日期),Linux 下的 Crontab 命令.同时,它还支持异步执行.后台执行调度任务. 一.基本架构 触发器 triggers:设定触发任务的条件 描述一个任务何时被触发,按日期或按时间间隔或按 cronjob 表达式三种方式触发 任务存储器 job stores:存放任务,可以放内存(默认)或数据库 注:调度器之间不能共享任务存储器 执行器 executors:用于执行任务,可设定执行模式 将指定的作业提交到线程池或者进程

  • 详解Python 定时框架 Apscheduler原理及安装过程

    在我们的日常工作自动化测试当中,几乎超过一半的功能都需要利用定时的任务来推动触发,例如在我们项目中有一个定时监控模块,根据自己设置的频率定时跑测试用例,定时检测是否存在线上紧急任务等等,这些都涉及到了有关定时任务的问题,很多情况下,大多数人会选择window的任务计划程序,但如果程序不在window平台下运行,就不能定时启动了:当然也可利用time模块的time.sleep()方法使程序休眠来达到定时任务的目的,但定时任务多了,代码可能看起来不太那么友好且有很大的局限性,因此,此时的 Apsch

  • Python中定时任务框架APScheduler的快速入门指南

    前言 大家应该都知道在编程语言中,定时任务是常用的一种调度形式,在Python中也涌现了非常多的调度模块,本文将简要介绍APScheduler的基本使用方法. 一.APScheduler介绍 APScheduler是基于Quartz的一个python定时任务框架,实现了Quartz的所有功能,使用起来十分方便.提供了基于日期.固定时间间隔以及crontab类型的任务,并且可以持久化任务. APScheduler提供了多种不同的调度器,方便开发者根据自己的实际需要进行使用:同时也提供了不同的存储机

  • python编写网页爬虫脚本并实现APScheduler调度

    前段时间自学了python,作为新手就想着自己写个东西能练习一下,了解到python编写爬虫脚本非常方便,且最近又学习了MongoDB相关的知识,万事具备只欠东风. 程序的需求是这样的,爬虫爬的页面是京东的电子书网站页面,每天会更新一些免费的电子书,爬虫会把每天更新的免费的书名以第一时间通过邮件发给我,通知我去下载. 一.编写思路: 1.爬虫脚本获取当日免费书籍信息 2.把获取到的书籍信息与数据库中的已有信息作比较,如果书籍存在不做任何操作,书籍不存在,执行插入数据库的操作,把数据的信息存入Mo

  • Python使用APScheduler实现定时任务过程解析

    前言 APScheduler是基于Quartz的一个Python定时任务框架.提供了基于日期.固定时间间隔以及crontab类型的任务,并且可以持久化任务. 在线文档:https://apscheduler.readthedocs.io/en/latest/userguide.html 一.安装APScheduler pip install apscheduler 二.基本概念 APScheduler有四大组件: 1.触发器 triggers : 触发器包含调度逻辑.每个作业都有自己的触发器,用

  • Python selenium爬虫实现定时任务过程解析

    现在需要启动一个selenium的爬虫,使用火狐驱动+多线程,大家都明白的,现在电脑管家显示CPU占用率20%,启动selenium后不停的开启浏览器+多线程, 好,没过5分钟,CPU占用率直接拉到90%+,电脑卡到飞起,定时程序虽然还在运行,但是已经类似于待机状态, 是不是突然感觉到面对电脑卡死,第一反应:卧槽,这个lj电脑,这么程序都跑不起来,我还写这么多代码,*****!! 是吧,接下来上代码,具体功能,请自信查阅相关资料深造: from datetime import datetime

  • python打包成so文件过程解析

    这篇文章主要介绍了python打包成so文件过程解析,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下 wget https://bootstrap.pypa.io/get-pip.py python get-pip.py pip install cython 编写setput.py文件: setup.py文件内容如下: from distutils.core import setup from distutils.extension import

  • python文字转语音实现过程解析

    这篇文章主要介绍了python文字转语音实现过程解析,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下 使用百度接口 接口地址 https://ai.baidu.com/docs#/TTS-Online-Python-SDK/top 安装接口 pip install baidu-aip from aip import AipSpeech """ 你的 APPID AK SK """ APP_ID =

  • SpringBoot基于数据库实现定时任务过程解析

    这篇文章主要介绍了SpringBoot基于数据库实现定时任务过程解析,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下 在我们平时开发的项目中,定时任务基本属于必不可少的功能,那大家都是怎么做的呢?但我知道的大多都是静态定时任务实现. 基于注解来创建定时任务非常简单,只需几行代码便可完成.实现如下: @Configuration @EnableScheduling public class SimpleScheduleTask { //10秒钟执行

  • python使用rsa非对称加密过程解析

    这篇文章主要介绍了python使用rsa非对称加密过程解析,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下 1.安装rsa 支持python 2.7 或者 python 3.5 以上版本 使用豆瓣pypi源来安装rsa pip install -i https://pypi.douban.com/simple rsa 2.加密解密 2.1.生成公私钥对 import rsa # 1.接收者(A)生成512位公私钥对 # a. lemon_pub为

  • Python测试线程应用程序过程解析

    这篇文章主要介绍了Python测试线程应用程序过程解析,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下 在本章中,我们将学习线程应用程序的测试.我们还将了解测试的重要性. 为什么要测试? 在我们深入讨论测试的重要性之前,我们需要知道测试的内容.一般来说,测试是一种了解某些东西是如何运作的技术.另一方面,特别是如果我们谈论计算机程序或软件,那么测试就是访问软件程序功能的技术. 在本节中,我们将讨论软件测试的重要性.在软件开发中,必须在向客户端发布软

  • python全局变量引用与修改过程解析

    这篇文章主要介绍了python全局变量引用与修改过程解析,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下 一.引用 使用到的全局变量只是作为引用,不在函数中修改它的值的话,不需要加global关键字.如: #! /usr/bin/python a = 1 b = [2, 3] def func(): if a == 1: print("a: %d" %a) for i in range(4): if i in b: print(&quo

  • Python namedtuple命名元组实现过程解析

    这篇文章主要介绍了Python namedtuple命名元组实现过程解析,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下 命名元组(namedtuple)是一种带有属性的元组,它们是组合只读数据的很好的方式. 相比一般的元组,构造命名元组需要先导入namedtuple,因为它不在默认的命名空间里.然后通过名字和属性来定义一个命名元组.这会返回一个像类一样的对象,可以进行多次实例化. 命名元组可以被打包.解包以及做所有可以对普通元组做的事,并且还可

  • python自动化unittest yaml使用过程解析

    这篇文章主要介绍了python自动化unittest yaml使用过程解析,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下 在编写unittest自动化用例时,一个请求需要编写多条用例,而涉及的参数基本相同,这时候就会用到配置文件,可以把参数配置项统一管理,避免重复代码,也方便后期维护 此处用到的是yaml,首先需要安装yaml库,pip install yaml 安装成功后,脚本导入语句,import yaml,具体语法可参照如上入门教程 举例

随机推荐