flask 架構下定時任務通常使用 flask_apscheduler 第三方庫。flask_apscheduler 功能非常強大，能滿足各種定時任務的需求。
安裝
pip3 install flask_apscheduler
使用方法
1.下面這段 example 是官網上的，我加了些注釋，友善學習。
from flask import Flask
from flask_apscheduler import APScheduler
class Config(object):
JOBS = [
{
'id': 'job1', # 一個辨別
'func': '__main__:job1', # 指定運作的函數
'args': (1, 2), # 傳入函數的參數
'trigger': 'interval', # 指定 定時任務的類型
'seconds': 10 # 運作的間隔時間
}
]
SCHEDULER_API_ENABLED = True
def job1(a, b): # 運作的定時任務的函數
print(str(a) + ' ' + str(b))
if __name__ == '__main__':
app = Flask(__name__) # 執行個體化flask
app.config.from_object(Config()) # 為執行個體化的 flask 引入配置
scheduler = APScheduler() # 執行個體化 APScheduler
# it is also possible to enable the API directly
# scheduler.api_enabled = True
scheduler.init_app(app) # 把任務清單放入 flask
scheduler.start() # 啟動任務清單
app.run() # 啟動 flask 2.多個定時任務
Config 類的屬性 JOBS 是清單，可以定義多個定時任務，如下
from flask import Flask
from flask_apscheduler import APScheduler
class Config(object):
JOBS = [
{
'id': '1',
'func': '__main__:job1',
'args': (1, 2),
'trigger': 'interval',
'seconds': 10
},
{
'id': '2',
'func': '__main__:job2',
'trigger': 'interval',
'seconds': 10
}
]
SCHEDULER_API_ENABLED = True
def job1(a, b):
print(str(a) + ' ' + str(b))
def job2():
print('Hello world!')
if __name__ == '__main__':
app = Flask(__name__)
app.config.from_object(Config())
scheduler = APScheduler()
# it is also possible to enable the API directly
# scheduler.api_enabled = True
scheduler.init_app(app)
scheduler.start()
app.run() View Code
觸發器類型
當你開始定時任務時，需要為定時政策選擇一個觸發器（設定 class Config 中 trigger 的值）。flask_apscheduler 提供了三種類型的觸發器。
- date 一次性指定固定時間，隻執行一次
- interval 間隔排程，隔多長時間執行一次
- cron 指定相對時間執行，比如：每月1号、每星期一執行
一、date 最基本的一種排程，指定固定時間，隻執行一次
- run_date（str）– 精确時間
class Config(object):
JOBS = [
{
'id': 'job1',
'func': '__main__:job1',
'args': (1, 2),
'trigger': 'date', # 指定任務觸發器 date
'run_date': '2020-7-23 16:50:00' # 指定時間 2020-7-23 16:50:00 執行
}
]
SCHEDULER_API_ENABLED = True 二、interval 通過設定 時間間隔 來運作定時任務
- weeks (int) – 間隔幾周
- days (int) – 間隔幾天
- hours (int) – 間隔幾小時
- minutes (int) – 間隔幾分鐘
- seconds (int) – 間隔多少秒
- start_date (datetime|str) – 開始日期
- end_date (datetime|str) – 結束日期
class Config(object):
JOBS = [
{
'id': 'job1',
'func': '__main__:job1',
'args': (1, 2),
'trigger': 'interval', # 指定任務觸發器 interval
'hours': 5 # 每間隔5h執行
}
]
SCHEDULER_API_ENABLED = True 三 、cron 通過設定 相對時間 來運作定時任務
- year (int|str) – 年，4位數字
- month (int|str) – 月 (範圍1-12)
- day (int|str) – 日 (範圍1-31)
- week (int|str) – 周 (範圍1-53)
- day_of_week (int|str) – 周内第幾天或者星期幾 (範圍0-6 或者 mon,tue,wed,thu,fri,sat,sun)
- hour (int|str) – 時 (範圍0-23)
- minute (int|str) – 分 (範圍0-59)
- second (int|str) – 秒 (範圍0-59)
- start_date (datetime|str) – 最早開始日期(包含)
- end_date (datetime|str) – 最晚結束時間(包含)
class Config(object):
JOBS = [
{
'id': 'job1',
'func': '__main__:job1',
'args': (1, 2),
'trigger': 'cron', # 指定任務觸發器 cron
'day_of_week': 'mon-fri', # 每周1至周5早上6點執行
'hour': 6,
'minute': 00
}
]
SCHEDULER_API_ENABLED = True 設定時區
開發好定時腳本後部署到 K8S，發現報錯 Timezone offset does not match system offset: 0 != 28800. Please, check your config files.
本地運作好好的，為什麼伺服器上就報錯了呢？
原因是系統的時區和代碼運作的時區不一緻導緻的。解決方法是在初始化 APScheduler() 的時候加上時區：BackgroundScheduler(timezone="Asia/Shanghai")
'''...'''
from apscheduler.schedulers.background import BackgroundScheduler
class config:
JOBS = ['''...''']
SCHEDULER_TIMEZONE = 'Asia/Shanghai'
'''...'''
if __name__ == '__main__':
scheduler = APScheduler(BackgroundScheduler(timezone="Asia/Shanghai"))
scheduler.init_app(app)
scheduler.start()
app.run()