专栏首页极客猴Python定时任务(下)

Python定时任务(下)

Photo from Unsplash

上篇文章,我们了解到有三种办法能实现定时任务,但是都无法做到循环执行定时任务。因此,需要一个能够担当此重任的库。它就是 APScheduler

1 简介

APScheduler 的全称是 Advanced Python Scheduler 。它是一个轻量级的 Python 定时任务调度框架。APScheduler 支持三种调度任务:固定时间间隔固定时间点(日期)Linux 下的 Crontab 命令。同时,它还支持异步执行、后台执行调度任务。

2 安装

使用 pip 包管理工具安装 APScheduler 是最方便快捷的。

pip install APScheduler
# 如果出现因下载失败导致安装不上的情况,建议使用代理
pip --proxy http://代理ip:端口 install APScheduler

3 使用步骤

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

1) 新建一个 schedulers (调度器) 。

2) 添加一个调度任务(job stores)。

3) 运行调度任务。

下面是执行每 2 秒报时的简单示例代码:

import datetime
import time
from apscheduler.schedulers.background import BackgroundScheduler

def timedTask():
    print(datetime.datetime.utcnow().strftime("%Y-%m-%d %H:%M:%S.%f")[:-3])

if __name__ == '__main__':
    # 创建后台执行的 schedulers
    scheduler = BackgroundScheduler()  
    # 添加调度任务
    # 调度方法为 timedTask,触发器选择 interval(间隔性),间隔时长为 2 秒
    scheduler.add_job(timedTask, 'interval', seconds=2)
    # 启动调度任务
    scheduler.start()
    
    while True:
        print(time.time())
        time.sleep(5)

4 基础组件

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

  • schedulers(调度器)

它是任务调度器,属于控制器角色。它配置作业存储器和执行器可以在调度器中完成,例如添加、修改和移除作业。

  • triggers(触发器)

描述调度任务被触发的条件。不过触发器完全是无状态的。

  • job stores(作业存储器)

任务持久化仓库,默认保存任务在内存中,也可将任务保存都各种数据库中,任务中的数据序列化后保存到持久化数据库,从数据库加载后又反序列化。

  • executors(执行器)

负责处理作业的运行,它们通常通过在作业中提交指定的可调用对象到一个线程或者进城池来进行。当作业完成时,执行器将会通知调度器。

4.1 schedulers(调度器)

我个人觉得 APScheduler 非常好用的原因。它提供 7 种调度器,能够满足我们各种场景的需要。例如:后台执行某个操作,异步执行操作等。调度器分别是:

  • BlockingScheduler :调度器在当前进程的主线程中运行,也就是会阻塞当前线程。
  • BackgroundScheduler :调度器在后台线程中运行,不会阻塞当前线程。
  • AsyncIOScheduler :结合 asyncio 模块(一个异步框架)一起使用。
  • GeventScheduler : 程序中使用 gevent(高性能的Python并发框架)作为 IO 模型,和 GeventExecutor 配合使用。
  • TornadoScheduler :程序中使用 Tornado(一个web框架)的IO模型,用 ioloop.add_timeout 完成定时唤醒。
  • TwistedScheduler : 配合 TwistedExecutor,用 reactor.callLater 完成定时唤醒。
  • QtScheduler : 你的应用是一个 Qt 应用,需使用 QTimer 完成定时唤醒。

4.2 triggers(触发器)

APScheduler 有三种内建的 trigger:

1)date 触发器

date 是最基本的一种调度,作业任务只会执行一次。它表示特定的时间点触发。它的参数如下:

date 触发器使用示例如下:

from datetime import datetime
from datetime import date
from apscheduler.schedulers.background import BackgroundScheduler

def 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()

2)interval 触发器

固定时间间隔触发。interval 间隔调度,参数如下:

interval 触发器使用示例如下:

import datetime
from apscheduler.schedulers.background import BackgroundScheduler

def 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()

3)cron 触发器

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

我们先了解 cron 参数:

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

cron 触发器使用示例如下:

import datetime
from apscheduler.schedulers.background import BackgroundScheduler

def 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()

4.3 作业存储(job store)

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

1)添加 job

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

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

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

import datetime
from 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()

2)移除 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()

3)获取 job 列表

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

4) 修改 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)

5)关闭 job

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

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

4.4 执行器(executor)

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

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

注:本代码来源于网络。

from pymongo import MongoClient
from apscheduler.schedulers.blocking import BlockingScheduler
from apscheduler.jobstores.mongodb import MongoDBJobStore
from apscheduler.jobstores.memory import MemoryJobStore
from apscheduler.executors.pool import ThreadPoolExecutor, ProcessPoolExecutor
 
 
def my_job():
    print 'hello world'
host = '127.0.0.1'
port = 27017
client = 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()

END

作者:猴哥

公众号:极客猴

爱好读书,喜欢钻研技术,梦想成为文艺青年的 boy。

本文分享自微信公众号 - 极客猴(Geek_monkey),作者:猴哥

原文出处及转载信息见文内详细说明,如有侵权,请联系 yunjia_community@tencent.com 删除。

原始发表时间:2017-12-13

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

我来说两句

0 条评论
登录 后参与评论

相关文章

  • 学爬虫之道

    Django 已经算是入门,所以自己把学习目标转到爬虫。自己接下来会利用三个月的时间来专攻 Python 爬虫。这几天,我使用“主题阅读方法”阅读 Python...

    猴哥yuri
  • Django 学习笔记之使用旧数据库

    如果你按照顺序,从第一篇文章读到本文。恭喜你,你已经将 Django 大部分基础知识掌握了。后续的文章是在之前的基础上添砖加瓦或常用的应用。本文将的内容是一个场...

    猴哥yuri
  • 你还在为 import 库而烦恼吗?试试这个库

    还记得入门Python数据分析时经常会import几个库,下面这几个可谓是入门学习时的四大护法,Python数据处理和可视化常会用的工具。

    猴哥yuri
  • 用数据分析告诉你数据分析师能挣多少钱

    随着大数据时代的到来和数据的市场价值得到认可,数据分析师、进阶一点的还有数据挖掘工程师、甚至是金字塔顶尖的数据科学家,这些作为21世纪最性感的职业已成功吸引无数...

    华章科技
  • Oracle Job创建及使用详解

    Oracle job有定时执行的功能,可以在指定的时间点或每天的某个时间点自行执行任务。 一、查询系统中的job,可以查询视图 --相关视图 select *...

    流浪的猫666
  • Python-jenkins模块获取jobs的执行状态操作

    在job执行结束前使用server_1.get_build_console_output(‘2019/get_node_list’,7).split(‘\n’)...

    砸漏
  • Python定时任务APScheduler原理及实例解析

    1、 线程睡眠函数 sleep() ——粗暴!一直占有 CPU 资源,导致后续操作无法执行

    砸漏
  • python apscheduler job处理

    scheduler.add_job(job_func, 'interval', minutes=2, id='job_one') scheduler.remo...

    用户5760343
  • 爬 Boss 直聘,分析 Python 工作现状

    要说在当今的编程圈,找10位程序猿询问下当前世界上最好的语言是哪个,那必须是 PHP(强迫症)!但是如果你询问当今最火爆的语言是哪个,那么80%的小伙伴儿会毫不...

    周萝卜
  • 左手用R右手Python系列——多进程/线程数据抓取与网页请求

    这一篇涉及到如何在网页请求环节使用多进程任务处理功能,因为网页请求涉及到两个重要问题:一是多进程的并发操作会面临更大的反爬风险,所以面临更严峻的反爬风险,二是抓...

    数据小磨坊

扫码关注云+社区

领取腾讯云代金券