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()

觸發器型別

當你開始定時任務時，需要為定時策略選擇一個觸發器（設定 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()