摘要：
Flask通常用于Flask框架下的调度任务_ apscheduler的第三方库。flask_Appscheduler非常强大，可以满足各种预定任务的需求。安装pip3installflare_如何使用apscheduler 1。以下示例在官方网站上。我添加了一些注释以方便学习。”）如果__name__=='__main__'：app=Flaskapp.config。from_objectscheduler=APScheduler（）#可以直接启用API#调度程序。api_enabled=Truescheduler。init_appscheduler.start（）应用程序。运行（）ViewCode触发器类型启动计划任务时，需要为计时策略选择触发器。flask_Appscheduler提供了三种类型的触发器。
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()