本篇文章給大家分享的是有關(guān)如何在python 項(xiàng)目中利用Apscheduler實(shí)現(xiàn)一個(gè)定時(shí)任務(wù)，小編覺得挺實(shí)用的，因此分享給大家學(xué)習(xí)，希望大家閱讀完這篇文章后可以有所收獲，話不多說，跟著小編一起來看看吧。
Trigger觸發(fā)器
Job作業(yè)
Excutor執(zhí)行器
Scheduler調(diào)度器
大概了解了Apscheduler包含的幾種概念，現(xiàn)在先來看一下一個(gè)簡(jiǎn)單的示例：
重慶分公司
重慶分公司
# -*- 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 ..........
這個(gè)簡(jiǎn)單的示例，我們用上面提到幾種組件分析一下運(yùn)行邏輯：
首先是Scheduler調(diào)度器，這個(gè)示例使用的BlockingScheduler調(diào)度器，在官方文檔中的解釋是，BlockingScheduler適合當(dāng)你的這個(gè)定時(shí)任務(wù)程序是運(yùn)行的程序；換言之，則是BlockingScheduler調(diào)度器是一個(gè)阻塞調(diào)度器，當(dāng)程序運(yùn)行這種調(diào)度器，進(jìn)程則會(huì)阻塞，無法執(zhí)行其他操作；
其次是Job作業(yè)和觸發(fā)器，這兩個(gè)放在一起講是因?yàn)椋诙x作業(yè)的時(shí)候，你就需要選擇一個(gè)觸發(fā)器，這里選擇的是interval觸發(fā)器，這種觸發(fā)器會(huì)以固定時(shí)間間隔運(yùn)行作業(yè)。換言之，為調(diào)度器添加一個(gè)hello的工作，并以每5秒的時(shí)間間隔執(zhí)行任務(wù)。
最后就是執(zhí)行器，默認(rèn)是ThreadPoolExcutor執(zhí)行器，他們將任務(wù)中可調(diào)用對(duì)象交給線程池執(zhí)行操作，等完成操作后，執(zhí)行器會(huì)通知調(diào)度程序。
內(nèi)置的三種Trigger觸發(fā)器類型：
date:特定時(shí)間僅運(yùn)行一次作業(yè)
interval: 固定的時(shí)間間隔內(nèi)運(yùn)行一次作業(yè)
cron: 在一天內(nèi)特定的時(shí)間定期運(yùn)行作業(yè)
常見的Scheduler調(diào)度器:
BlockingScheduler: 調(diào)度程序是流程中運(yùn)行的東西
BackgroundScheduler: 調(diào)度程序在應(yīng)用程序內(nèi)部的后臺(tái)運(yùn)行時(shí)使用
AsyncIOScheduler: 應(yīng)用程序使用asyncio模塊
GeventScheduler: 應(yīng)用程序使用gevent模塊
TornadoScheduler：構(gòu)建Tornado應(yīng)用程序時(shí)使用
TwistedScheduler: 構(gòu)建Tornado應(yīng)用程序時(shí)使用
QtScheduler: 在構(gòu)建QT應(yīng)用程序時(shí)使用
常見的JobStore：
MemoryJobStore
MongoDBJobStore
SQLAlchemyJobStore
RedisJobStore
通過上面一個(gè)簡(jiǎn)單的示例了解大概的工作流程，以及各個(gè)組件在整個(gè)流程中的作用，以下的示例是Flask Web框架結(jié)合使用Apscheduler定時(shí)器，定時(shí)執(zhí)行任務(wù)。
# -*- 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，將任務(wù)序列化存入到Redis數(shù)據(jù)庫(kù)中。這里順便提一下，為什么需要設(shè)置作業(yè)存儲(chǔ)器，原因是當(dāng)調(diào)度器程序崩潰時(shí)，仍然能夠保留作業(yè)，當(dāng)然選擇什么作業(yè)存儲(chǔ)器，可以根據(jù)具體的工作場(chǎng)景，目前主流的mysql，mongodb，redis，SQLite基本都支持;
然后再看看Scheduler，這里使用的時(shí)BackgroundScheduler，因?yàn)檫@里要求調(diào)度程序不能阻塞flask程序的正常接收請(qǐng)求，所以選在BackgrounScheduler讓它在開始執(zhí)行任務(wù)時(shí)是在后臺(tái)運(yùn)行的，不會(huì)阻塞主線程;
最后看看工作的邏輯，這里get_job獲取作業(yè)的狀態(tài)，查看作業(yè)是否存在，start_job則是先判斷作業(yè)是否啟動(dòng)，然后再?zèng)Q定啟動(dòng)操作，remove_job則是停止作業(yè)。而這里的作業(yè)定義則是通過interval觸發(fā)器，每五秒執(zhí)行一次say_hello任務(wù)；
以上就是如何在python 項(xiàng)目中利用Apscheduler實(shí)現(xiàn)一個(gè)定時(shí)任務(wù)，小編相信有部分知識(shí)點(diǎn)可能是我們?nèi)粘９ぷ鲿?huì)見到或用到的。希望你能通過這篇文章學(xué)到更多知識(shí)。更多詳情敬請(qǐng)關(guān)注創(chuàng)新互聯(lián)行業(yè)資訊頻道。
