導(dǎo)語(yǔ)

在工作場(chǎng)景遇到了這么一個(gè)場(chǎng)景，就是需要定期去執(zhí)行一個(gè)緩存接口，用于同步設(shè)備配置。首先想到的就是Linux上的crontab，可以定期，或者間隔一段時(shí)間去執(zhí)行任務(wù)。但是如果你想要把這個(gè)定時(shí)任務(wù)作為一個(gè)模塊集成到Python項(xiàng)目中，或者想持久化任務(wù)，顯然crontab不太適用。Python的APScheduler模塊能夠很好的解決此類問(wèn)題，所以專門寫這篇文章，從簡(jiǎn)單入門開(kāi)始記錄關(guān)于APScheduler最基礎(chǔ)的使用場(chǎng)景，以及解決持久化任務(wù)的問(wèn)題，最后結(jié)合其他框架深層次定制定時(shí)任務(wù)模塊這幾個(gè)點(diǎn)入手。

簡(jiǎn)單介紹

先簡(jiǎn)單介紹一下Apscheduler模塊包含的四種組件：

• Trigger觸發(fā)器
• Job作業(yè)
• Excutor執(zhí)行器
• Scheduler調(diào)度器
  

大概了解了Apscheduler包含的幾種概念，現(xiàn)在先來(lái)看一下一個(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ì)阻塞，無(wú)法執(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è)
  

常見(jiàn)的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í)使用
  

常見(jiàn)的JobStore：

• MemoryJobStore
• MongoDBJobStore
• SQLAlchemyJobStore
• RedisJobStore
  

進(jìn)階使用

通過(guò)上面一個(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讓它在開(kāi)始執(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è)定義則是通過(guò)interval觸發(fā)器，每五秒執(zhí)行一次say_hello任務(wù)；

總結(jié)

最后總結(jié)一下，首先你要設(shè)置一個(gè)作業(yè)存儲(chǔ)器用于在調(diào)度程序崩潰重新恢復(fù)時(shí)，還能夠在作業(yè)存儲(chǔ)器中獲取到作業(yè)繼續(xù)執(zhí)行；然后你需要設(shè)置一個(gè)執(zhí)行器，這個(gè)根據(jù)作業(yè)的類型，比如時(shí)一個(gè)CPU密集型的任務(wù)，那就可以用進(jìn)程池執(zhí)行器，默認(rèn)是用線程池執(zhí)行器；最后創(chuàng)建配置調(diào)度器，啟動(dòng)調(diào)度，可以在啟動(dòng)前添加作業(yè)，也可以在啟動(dòng)后添加，刪除，獲取作業(yè)。(在這里需要明白的一點(diǎn)就是應(yīng)用程序不會(huì)直接去操作作業(yè)存儲(chǔ)器，作業(yè)或者執(zhí)行器，而是調(diào)度器提供適當(dāng)?shù)慕涌趤?lái)處理這些接口。)

ApScheduler是一個(gè)不錯(cuò)的定時(shí)任務(wù)庫(kù)，能夠動(dòng)態(tài)的添加刪除，同時(shí)也支持不同的觸發(fā)器類型，這也是它的優(yōu)勢(shì)，相反一些如果是靜態(tài)任務(wù)，其實(shí)可以用如linux的crontab工具去做定時(shí)任務(wù)。有關(guān)這方面的記錄還會(huì)持續(xù)更新，如果有什么問(wèn)題，可以提出來(lái)，大家一起探討。

以上就是python Apscheduler的使用方法的詳細(xì)內(nèi)容，更多關(guān)于python Apscheduler的資料請(qǐng)關(guān)注腳本之家其它相關(guān)文章！

最新評(píng)論