Background

本篇是記錄一下 APScheduler 的使用方式 。 那大概講一下使用背景,我自己本身在使用 Flask 框架,那剛好有需求需要執行類似cronjob的任務來判斷任務執行時間是否到期,那以前的我都是另外再開一個專案專門寫類似這種cronjob的程式,但最近突然靈機一動不知道有沒有第三方的套件可以幫助我完成這件事情,一查之後發現了 APScheduler 這個套件。

Installation

$ pip3 install apscheduler

Basic concepts

APScheduler 共有四個主要元素所組成

  • triggers

triggers 包含 scheduling logic。每個作業項目有各自的 trigger 用於確定何時應該運行作業。除了初始配置之外,trigger 是完全無狀態的。

  • job stores

job stores 安排預定的工作。初始的 job stores 只是將作業保留在記憶體中,但亦可跟各式的資料庫做結合。作業的數據在保存到持久性作業存儲時會被序列化,並在從其中加載時進行反序列化。作業存儲(Default初始值之外)不會將作業數據保留在內存中,而是充當中間人,用於在後端保存,加載,更新和搜索作業。schedulers是不能共享job stores的。

  • executors

executors 是用來處理作業的運行。他們通常通過將作業中指定的可調用對象提交給線程(thread)或進程(process)池來完成此操作。作業完成後,executors 會通知 schedulers。

  • schedulers

schedulers 則是把上述三項整合在一起。您通常只在應用程序中運行一個 schedulers , 應用程序開發人員通常不直接處理 job stores,executors 或 triggers。相反,schedulers 提供了適當的接口來處理所有這些東西。配置 job stores 和 executors 是通過 schedulers 完成的,添加,修改和刪除作業也是如此。

APScheduler提供了多種調度器,可以根據具體需求來選擇合適的調度器,常用的調度器有:

  • BlockingScheduler:適合於只在進程中運行單個任務的情況,通常在調度器是你唯一要運行的東西時使用。

  • BackgroundScheduler: 適合於要求任何在程序在背景模式運行的情況,當希望調度器在應用背景模式執行時使用。

  • AsyncIOScheduler:適合於使用asyncio框架的情況

  • GeventScheduler: 適合於使用gevent框架的情況

  • TornadoScheduler: 適合於使用Tornado框架的應用

  • TwistedScheduler: 適合使用Twisted框架的應用

APScheduler comes with three built-in trigger types:

  • date: use when you want to run the job just once at a certain point of time
  • interval: use when you want to run the job at fixed intervals of time
  • cron: use when you want to run the job periodically at certain time(s) of day

Starting the scheduler

那這邊只會紀錄最簡單 Default 的使用方式,如果對其他更深入的東西有興趣的話可以到官方文件去看相關使用方式。

我自己有使用 Flask 的框架,所以接下來的範例我會把 scheduler & Flask 整合在一起使用。

$ mkdir APSscheduler
$ cd APSscheduler
$ virtualenv .venv
$ source .venv/bin/activate
$ touch run.py cronjob.py
$ pip3 install Flask
$ pip3 install apscheduler

run.py

from flask import Flask
from flask import jsonify
import sys
import os
from cronjob import hello
from cronjob import scheduler

app = Flask(__name__)
scheduler.add_job(hello, "interval", minutes=1)
scheduler.start()

if __name__ == "__main__":
    app.run(use_reloader=False)

我這邊是設定每一分鐘跑一次 hello 這個function

基本參數如下:

  • year (int | str) - 年 (ex: 2019)
  • month (int | str) - 月(範圍1-12)
  • day (int | str) - 日(範圍1-31)
  • week (int | str) - 週(範圍1-53)
  • hour (int | str) - 時(範圍0-23)
  • minute(int | str) - 分(範圍0-59)
  • second(int | str) - 秒(範圍0-59)
  • day_of_week(int | str)- 星期幾(範圍0-6或者mon,tue,wed,thu,fri,sat,sun)

cronjob.py

from apscheduler.schedulers.background import BackgroundScheduler
from datetime import datetime
scheduler = BackgroundScheduler(daemon=True)

def hello():
    print("hello world")

以下是執行結果

$ python3 run.py
 * Serving Flask app "run" (lazy loading)
 * Environment: production
   WARNING: This is a development server. Do not use it in a production deployment.
   Use a production WSGI server instead.
 * Debug mode: off
 * Running on http://127.0.0.1:5000/ (Press CTRL+C to quit)
hello world
hello world
hello world
hello world

Reference