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