欢迎您访问程序员文章站本站旨在为大家提供分享程序员计算机编程知识!
您现在的位置是: 首页  >  IT编程

python 基于Apscheduler实现定时任务

程序员文章站 2022-06-24 17:06:21
导语在工作场景遇到了这么一个场景,就是需要定期去执行一个缓存接口,用于同步设备配置。首先想到的就是linux上的crontab,可以定期,或者间隔一段时间去执行任务。但是如果你想要把这个定时任务作为一...

导语

在工作场景遇到了这么一个场景,就是需要定期去执行一个缓存接口,用于同步设备配置。首先想到的就是linux上的crontab,可以定期,或者间隔一段时间去执行任务。但是如果你想要把这个定时任务作为一个模块集成到python项目中,或者想持久化任务,显然crontab不太适用。python的apscheduler模块能够很好的解决此类问题,所以专门写这篇文章,从简单入门开始记录关于apscheduler最基础的使用场景,以及解决持久化任务的问题,最后结合其他框架深层次定制定时任务模块这几个点入手。

简单介绍

先简单介绍一下apscheduler模块包含的四种组件:

  • trigger触发器
  • job作业
  • excutor执行器
  • scheduler调度器

大概了解了apscheduler包含的几种概念,现在先来看一下一个简单的示例:

# -*- coding: utf-8 -*-

from apscheduler.schedulers.blocking import blockingscheduler
import time


def hello():
  print(time.strftime("%c"))


if __name__ == "__main__":
  scheduler = blockingscheduler()
  scheduler.add_job(hello, 'interval', seconds=5)
  scheduler.start()

示例的输出:

thu dec 3 16:01:20 2020
thu dec 3 16:01:25 2020
thu dec 3 16:01:30 2020
thu dec 3 16:01:35 2020
thu dec 3 16:01:40 2020
..........

这个简单的示例,我们用上面提到几种组件分析一下运行逻辑:

  • 首先是scheduler调度器,这个示例使用的blockingscheduler调度器,在官方文档中的解释是,blockingscheduler适合当你的这个定时任务程序是唯一运行的程序;换言之,则是blockingscheduler调度器是一个阻塞调度器,当程序运行这种调度器,进程则会阻塞,无法执行其他操作;
  • 其次是job作业和触发器,这两个放在一起讲是因为,在定义作业的时候,你就需要选择一个触发器,这里选择的是interval触发器,这种触发器会以固定时间间隔运行作业。换言之,为调度器添加一个hello的工作,并以每5秒的时间间隔执行任务。
  • 最后就是执行器,默认是threadpoolexcutor执行器,他们将任务中可调用对象交给线程池执行操作,等完成操作后,执行器会通知调度程序。

内置的三种trigger触发器类型:

  • date:特定时间仅运行一次作业
  • interval: 固定的时间间隔内运行一次作业
  • cron: 在一天内特定的时间定期运行作业

常见的scheduler调度器:

  • blockingscheduler: 调度程序是流程中唯一运行的东西
  • backgroundscheduler: 调度程序在应用程序内部的后台运行时使用
  • asyncioscheduler: 应用程序使用asyncio模块
  • geventscheduler: 应用程序使用gevent模块
  • tornadoscheduler:构建tornado应用程序时使用
  • twistedscheduler: 构建tornado应用程序时使用
  • qtscheduler: 在构建qt应用程序时使用

常见的jobstore:

  • memoryjobstore
  • mongodbjobstore
  • sqlalchemyjobstore
  • redisjobstore

进阶使用

通过上面一个简单的示例了解大概的工作流程,以及各个组件在整个流程中的作用,以下的示例是flask web框架结合使用apscheduler定时器,定时执行任务。

# -*- coding: utf-8 -*-

from flask import flask, blueprint, request
from apscheduler.executors.pool import threadpoolexecutor 
from apscheduler.schedulers.background import backgroundscheduler
from apscheduler.jobstores.redis import redisjobstore
import time

app = flask(__name__)
executors = {"default": threadpoolexecutor(5)}
default_redis_jobstore = redisjobstore(db=2, 
    jobs_key="apschedulers.default_jobs",
    run_times_key="apschedulers.default_run_times",
    host = '127.0.0.1',
    port = 6379
    )

scheduler = backgroundscheduler(executors=executors)
scheduler.add_jobstore(default_redis_jobstore)
scheduler.start()

def say_hello():
  print(time.strftime("%c"))


@app.route("/get_job", methods=['get'])
def get_job():
  if scheduler.get_job("say_hello_test"):
    return "yes"
  else:
    return "no"

@app.route("/start_job", methods=["get"])
def start_job():
  if not scheduler.get_job("say_hello_test"):
    scheduler.add_job(say_hello, "interval", seconds=5, id="say_hello_test")
    return "start scuessfully!"
  else:
    return "started failed"
  
@app.route("/remove_job", methods=["get"])
def remove_job():
  if scheduler.get_job("say_hello_test"):
    scheduler.remove_job("say_hello_test")
    return "delete successfully!"
  else:
    return "delete failed"


if __name__ == "__main__":
  app.run(host="127.0.0.1", port=8787, debug=true)
  • 先分析jobstore,这里使用的是redisjobstore,将任务序列化存入到redis数据库中。这里顺便提一下,为什么需要设置作业存储器,原因是当调度器程序崩溃时,仍然能够保留作业,当然选择什么作业存储器,可以根据具体的工作场景,目前主流的mysql,mongodb,redis,sqlite基本都支持;
  • 然后再看看scheduler,这里使用的时backgroundscheduler,因为这里要求调度程序不能阻塞flask程序的正常接收请求,所以选在backgrounscheduler让它在开始执行任务时是在后台运行的,不会阻塞主线程;
  • 最后看看工作的逻辑,这里get_job获取作业的状态,查看作业是否存在,start_job则是先判断作业是否启动,然后再决定启动操作,remove_job则是停止作业。而这里的作业定义则是通过interval触发器,每五秒执行一次say_hello任务;

总结

最后总结一下,首先你要设置一个作业存储器用于在调度程序崩溃重新恢复时,还能够在作业存储器中获取到作业继续执行;然后你需要设置一个执行器,这个根据作业的类型,比如时一个cpu密集型的任务,那就可以用进程池执行器,默认是用线程池执行器;最后创建配置调度器,启动调度,可以在启动前添加作业,也可以在启动后添加,删除,获取作业。(在这里需要明白的一点就是应用程序不会直接去操作作业存储器,作业或者执行器,而是调度器提供适当的接口来处理这些接口。)

apscheduler是一个不错的定时任务库,能够动态的添加删除,同时也支持不同的触发器类型,这也是它的优势,相反一些如果是静态任务,其实可以用如linux的crontab工具去做定时任务。有关这方面的记录还会持续更新,如果有什么问题,可以提出来,大家一起探讨。

以上就是python apscheduler的使用方法的详细内容,更多关于python apscheduler的资料请关注其它相关文章!