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

前言

大家应该都知道在编程语言中,定时任务是常用的一种调度形式,在Python中也涌现了非常多的调度模块,本文将简要介绍APScheduler的基本使用方法。

一、APScheduler介绍

APScheduler是基于Quartz的一个python定时任务框架,实现了Quartz的所有功能,使用起来十分方便。提供了基于日期、固定时间间隔以及crontab类型的任务,并且可以持久化任务。

APScheduler提供了多种不同的调度器,方便开发者根据自己的实际需要进行使用;同时也提供了不同的存储机制,可以方便与Redis,数据库等第三方的外部持久化机制进行协同工作,总之功能非常强大和易用。

在Python的世界中,另外一个齐名的调度模块是Celery,功能也非常的强大,号称分布式的调度器,感兴趣的读者可以自行进行研究。

官网文档地址:http://apscheduler.readthedocs.io/en/latest/

安装包位置: https://pypi.python.org/pypi/APScheduler/

在系统中,如何进行安装呢?其实非常简单,基于pip直接安装即可: 

pip install APScheduler 

二、APScheduler的主要的调度类

在APScheduler中有以下几个非常重要的概念,需要大家理解:   

1、触发器(trigger)

  包含调度逻辑,每一个作业有它自己的触发器,用于决定接下来哪一个作业会运行,根据trigger中定义的时间点,频率,时间区间等等参数设置。除了他们自己初始配置以外,触发器完全是无状态的。

2、作业存储(job store)

  存储被调度的作业,默认的作业存储是简单地把作业保存在内存中,其他的作业存储是将作业保存在数据库中。一个作业的数据讲在保存在持久化作业存储时被序列化,并在加载时被反序列化。调度器不能分享同一个作业存储。job store支持主流的存储机制:redis, mongodb, 关系型数据库, 内存等等

3、执行器(executor)

  处理作业的运行,他们通常通过在作业中提交制定的可调用对象到一个线程或者进城池来进行。当作业完成时,执行器将会通知调度器。基于池化的操作,可以针对不同类型的作业任务,更为高效地使用cpu的计算资源。
调度器(scheduler)

  通常在应用只有一个调度器,调度器提供了处理这些的合适的接口。配置作业存储和执行器可以在调度器中完成,例如添加、修改和移除作业。

这里简单列一下常用的若干调度器:

  • BlockingScheduler:仅可用在当前你的进程之内,与当前的进行共享计算资源
  • BackgroundScheduler: 在后台运行调度,不影响当前的系统计算运行
  • AsyncIOScheduler: 如果当前系统中使用了async module,则需要使用异步的调度器
  • GeventScheduler: 如果使用了gevent,则需要使用该调度
  • TornadoScheduler: 如果使用了Tornado, 则使用当前的调度器
  • TwistedScheduler:Twister应用的调度器
  • QtScheduler: Qt的调度器

由此可知,在APscheduler的调度器中,是与底层的实现机制紧密相关的,需要依据当前的计算模型来动态选择调度器。

三、APScheduler的job管理

Job是APScheduler中的核心,其承接目前需要执行的工作和任务,其可以在系统运行过程中动态地进行增加/修改/删除/查询等操作。

3.1 Job的新增

共有两种方式进行新增job的操作:

基于add_job来动态增加

代码示例:

sched.add_job(job_function, 'cron', day_of_week='mon-fri', hour='0-9', minute="*", second="*/4") 

基于修饰器scheduled_job来动态装饰job的实际函数

代码示例:

@sched.scheduled_job('cron', id='my_job_id', day='last sun')
def some_decorated_task():
 print("I am printed at 00:00:00 on the last Sunday of every month!") 

3.2 移除作业

job = scheduler.add_job(myfunc, 'interval', minutes=2)
job.remove()
Same, using an explicit job ID: 

scheduler.add_job(myfunc, 'interval', minutes=2, id='my_job_id')
scheduler.remove_job('my_job_id') 

基于job id来动态移除特定的job.

3.3 暂停和恢复作业

暂停作业:

apscheduler.job.Job.pause()

apscheduler.schedulers.base.BaseScheduler.pause_job()

恢复作业:

apscheduler.job.Job.resume()

apscheduler.schedulers.base.BaseScheduler.resume_job()

3.4. 获得job列表

获得调度作业的列表,可以使用 get_jobs() 来完成,它会返回所有的job实例。或者使用 print_jobs() 来输出所有格式化的作业列表。

3.5. 修改作业 job

可以通过apscheduler.job.Job.modify() or modify_job()来动态修改job的属性信息,除了job id无法修改之外,都是可以修改的。

job.modify(max_instances=6, name='Alternate name') 

另外我们也可以通过apscheduler.job.Job.reschedule() or reschedule_job()动态重新设置trigger,示例如下:

scheduler.reschedule_job('my_job_id', trigger='cron', minute='*/5') 

 3.6. 关闭调度器

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

scheduler.shutdown()
scheduler.shutdown(wait=False) 

四、 APScheduler的代码示例

这里使用装饰器来展示一个调度的使用:

from apscheduler.schedulers.blocking import BlockingScheduler 

sched = BlockingScheduler() 

@sched.scheduled_job('interval', seconds=3)
def timed_job():
 print('This job is run every three minutes.') 

@sched.scheduled_job('cron', day_of_week='mon-fri', hour='0-9', minute='30-59', second='*/3')
def scheduled_job():
 print('This job is run every weekday at 5pm.') 

print('before the start funciton')
sched.start()
print("let us figure out the situation") 

代码说明:

在这段代码中,使用了当前进程中共享计算资源的BlockingScheduler,共使用了2个调度器,其中一个是间隔3秒的执行。

另外一个调度器是模仿cron来执行的,在周一到周五其间,每天的0点到9点直接,在30分到59分之间执行,执行频次为3秒。

基于正常代码的示例如下:

from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.schedulers.blocking import BlockingScheduler 

import datetime
import time
import logging 

def job_function():
 print "Hello World" + " " + str(datetime.datetime.now()) 

if __name__ == '__main__':
 log = logging.getLogger('apscheduler.executors.default')
 log.setLevel(logging.INFO) # DEBUG 

 fmt = logging.Formatter('%(levelname)s:%(name)s:%(message)s')
 h = logging.StreamHandler()
 h.setFormatter(fmt)
 log.addHandler(h) 

 print('start to do it') 

 sched = BlockingScheduler() 

 # Schedules job_function to be run on the third Friday
 # of June, July, August, November and December at 00:00, 01:00, 02:00 and 03:00
 sched.add_job(job_function, 'cron', day_of_week='mon-fri', hour='0-9', minute="*", second="*/4") 

 sched.start() 

五、某个异常问题的思考

在执行以下代码之时候,定时任务一直未能正常生效:

from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.schedulers.blocking import BlockingScheduler 

import datetime
import time 

def job_function():
 print "Hello World" + " " + str(datetime.datetime.now()) 

if __name__ == '__main__':
 print('start to do it')
 sched = BlockingScheduler()
 sched.add_job(job_function, 'cron', day_of_week='mon-fri', hour='0-9', minute="*", second="*/4")
 sched.start() 

代码报错的错误信息为:

No handlers could be found for logger “apscheduler.scheduler” 

从字面意思来分析,是没有logging模块的logger存在,故需要添加上去即可。

新增对应的logging信息即可:

import logging 
log = logging.getLogger('apscheduler.executors.default')
 log.setLevel(logging.INFO) # DEBUG 

 fmt = logging.Formatter('%(levelname)s:%(name)s:%(message)s')
 h = logging.StreamHandler()
 h.setFormatter(fmt)
 log.addHandler(h) 

后来笔者重新做了一次执行,即使移除掉logging的内容,依然可以正常执行,故可以推测为需要动态引入一次依赖包logging即可。

六、总结

APScheduler是一个非常强大易用的类库,为了我们简单快捷的解决问题提供了很多的工具,并且提供了很多灵活的扩展点,只要你添加若干的web页面,就可以创建一个强大的任务调度系统,不是吗?

好了,以上就是这篇文章的全部内容了,希望本文的内容对大家的学习或者工作能带来一定的帮助,如果有疑问大家可以留言交流,谢谢大家对我们的支持。

(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编写网页爬虫脚本并实现APScheduler调度

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

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

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

  • Python定时任务框架APScheduler原理及常用代码

    APScheduler简介 在平常的工作中几乎有一半的功能模块都需要定时任务来推动,例如项目中有一个定时统计程序,定时爬出网站的URL程序,定时检测钓鱼网站的程序等等,都涉及到了关于定时任务的问题,第一时间想到的是利用time模块的time.sleep()方法使程序休眠来达到定时任务的目的,虽然这样也可以,但是总觉得不是那么的专业,^_^所以就找到了python的定时任务模块APScheduler: APScheduler基于Quartz的一个Python定时任务框架,实现了Quartz的所有功

  • Python定时任务框架APScheduler安装使用详解

    目录 前言 一.APscheduler简介 二.APscheduler安装 三.APscheduler组成部分 1.Job 作业 2.Trigger 触发器 3.Jobstore 作业存储 4.Executor 执行器 5.scheduler 调度器 四.Scheduler工作流程图 1.Scheduler添加job流程 2.Scheduler调度流程 五.APscheduler使用 1.简单应用 2.操作作业 2.1 date触发器 2.2 interval触发器 2.3 cron触发器 参考

  • Python中Scrapy框架的入门教程分享

    目录 前言 安装Scrapy 创建一个Scrapy项目 创建一个爬虫 运行爬虫 结论 前言 Scrapy是一个基于Python的Web爬虫框架,可以快速方便地从互联网上获取数据并进行处理.它的设计思想是基于Twisted异步网络框架,可以同时处理多个请求,并且可以使用多种处理数据的方式,如提取数据.存储数据等. 本教程将介绍如何使用Scrapy框架来编写一个简单的爬虫,从而让您了解Scrapy框架的基本使用方法. 安装Scrapy 首先,您需要在您的计算机上安装Scrapy框架.您可以使用以下命

  • python中Flask框架简单入门实例

    本文实例讲述了python中Flask框架的简单用法.分享给大家供大家参考.具体如下: 使用Flask框架的简单入门范例代码,如果你正学习Flask框架,可以参考下面的启动代码,这段代码可以在网页上输出"hello world" import os # Using Flask since Python doesn't have built-in session management from flask import Flask, session app = Flask(__name_

  • MyBatis框架零基础快速入门案例详解

    目录 一.创建数据库和表 二.创建maven工程 三.代码编写 1.编写Student实体类 2.编写DAO接口StudentDao 3.编写DAO接口Mapper映射文件StudentDao.xml. 4.创建MyBatis主配置文件 四.创建测试类进行测试 1.创建测试类MyBatisTest 2.配置日志功能 五.增删改操作 insert操作 MyBatis下载地址:https://github.com/mybatis/mybatis-3/releases 一.创建数据库和表 数据库名ss

  • Python中Django框架利用url来控制登录的方法

    本文实例讲述了Python中Django框架利用url来控制登录的方法.分享给大家供大家参考.具体如下: from django.conf.urls.defaults import patterns,url #or use login_required from django.contrib.admin.views.decorators import staff_member_required def login_url(regex, view, *p,**args): """

  • python中django框架通过正则搜索页面上email地址的方法

    本文实例讲述了python中django框架通过正则搜索页面上email地址的方法.分享给大家供大家参考.具体实现方法如下: import re from django.shortcuts import render from pattern.web import URL, DOM, abs, find_urls def index(request): """ find email addresses in requested url or contact page &quo

  • 解决Python中定时任务线程无法自动退出的问题

    python的线程有一个类叫Timer可以,用来创建定时任务,但是它的问题是只能运行一次,如果要重复执行,则只能在任务中再调用一次timer,但这样就存在新的问题了,就是在主进程退出后,不能正常退出子线程. from threading import Timer def scheduletaskwrap(): pritn "in task" Timer(10, scheduletaskwrap).start() Timer(10, scheduletaskwrap).start() 象

  • python中的yield from语法快速学习

    协程是什么?可能很多人不清楚,所以我们先从其中的yield from开始讲解. yield from 用法详解: yield from 是在Python3.3才出现的语法.所以这个特性在Python2中是没有的. yield from 后面需要加的是可迭代对象,它可以是普通的可迭代对象,也可以是迭代器,甚至是生成器. 简单应用:拼接可迭代对象 我们可以用一个使用yield和一个使用yield from的例子来对比看下. 使用yield <p style="line-height: 1.75

随机推荐