如何在python 项目中利用Apscheduler实现一个定时任务

发布时间:2020-12-15 17:27:03 作者:Leah
来源:亿速云 阅读:307

本篇文章给大家分享的是有关如何在python 项目中利用Apscheduler实现一个定时任务,小编觉得挺实用的,因此分享给大家学习,希望大家阅读完这篇文章后可以有所收获,话不多说,跟着小编一起来看看吧。

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

大概了解了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
..........

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

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

常见的Scheduler调度器:

常见的JobStore:

进阶使用

通过上面一个简单的示例了解大概的工作流程,以及各个组件在整个流程中的作用,以下的示例是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)

以上就是如何在python 项目中利用Apscheduler实现一个定时任务,小编相信有部分知识点可能是我们日常工作会见到或用到的。希望你能通过这篇文章学到更多知识。更多详情敬请关注亿速云行业资讯频道。

推荐阅读:
  1. python GUI库图形界面开发之pyinstaller打包python程序为exe安装文件
  2. 如何在python中实现BlockingScheduler定时任务

免责声明:本站发布的内容(图片、视频和文字)以原创、转载和分享为主,文章观点不代表本网站立场,如果涉及侵权请联系站长邮箱:is@yisu.com进行举报,并提供相关证据,一经查实,将立刻删除涉嫌侵权内容。

python apscheduler

上一篇:利用C语言怎么编写一个扫雷小游戏

下一篇:利用go语言怎么查找出两个数组不同的地方

相关阅读

您好,登录后才能下订单哦!

密码登录
登录注册
其他方式登录
点击 登录注册 即表示同意《亿速云用户服务条款》