描述
APScheduler基于Quartz的一个Python定时任务框架，实现了Quartz的所有功能。最近使用Flask框架使用Flask_APScheduler来做定时任务，在使用过程当中也遇到很多问题，例如在定时任务调用的方法中需要用到flask的app.app_context()时，需要使用current_app记录日志时，例如：current_app.logger.info(“my_job已执行”)，定时任务中使用current_app对象会报错，查看了很多资料，大部分资料都是说没有app就创建一个，这样确实也能解决，但是我总感觉这种解决是有问题的,拿.Net Core来说，使用Quartz.NET定时任务时，定时任务依赖于一个Host(主机)对象，不需要重复创建Host对象，但是Flask的app对象使用过程中却需要重新create app，Quartz.NET也是基于Quartz的定时任务框架，我使用过Quartz.NET，因此始终觉得Flask_APScheduler中create app使用是有问题，终于在过了一段时间后看到一位前辈使用Flask_APScheduler的一篇文章后，瞬间通达了，这个问题终于得到完美解决
最佳使用Flask_APScheduler
安装Flask_APScheduler
pip install Flask_APScheduler1.项目结构图如下：
2.Python 软件包utils下的__init__.py 初始化生成APScheduler对象
这里可以灵活处理，例如：也可以是common软件包下__init__.py里初始化APScheduler
__init__.py的代码如下：
import atexit
import platform
from flask_apscheduler import APScheduler
# 初始化生成APScheduler对象
scheduler = APScheduler()
def init_scheduler(app):
# 解决APScheduler定时任务重复执行的问题
if platform.system() == 'Linux':
# Linux 环境下
fcntl = __import__("fcntl")
f = open('scheduler.lock', 'wb')
try:
fcntl.flock(f, fcntl.LOCK_EX | fcntl.LOCK_NB)
scheduler.init_app(app)
scheduler.start()
except Exception as e:
app.logger.error(e)
print(e)
def unlock():
fcntl.flock(f, fcntl.LOCK_UN)
f.close()
atexit.register(unlock)
else:
# Window 环境下
msvcrt = __import__('msvcrt')
f = open('scheduler.lock', 'wb')
try:
msvcrt.locking(f.fileno(), msvcrt.LK_NBLCK, 1)
scheduler.init_app(app)
scheduler.start()
except Exception as e:
pass
def _unlock_file():
try:
f.seek(0)
msvcrt.locking(f.fileno(), msvcrt.LK_UNLCK, 1)
except Exception as e:
pass
atexit.register(_unlock_file)3. config.py配置类代码
class Config:
JOBS = [
{
'id': 'job1',
'func': 'app:MyService.my_job', # 注意这里的格式，app 是 Flask 应用对象的名称(app.py)，: 后面是任务函数名
'kwargs': {'job_name': 'job1'},
'trigger': 'cron',
'hour': 16, # 16 点执行
'minute': 58, # 58 分执行
'second': 0 # 0 秒执行
},
{
'id': 'job2',
'func': 'app:MyService.my_job', # 注意这里的格式，app 是 Flask 应用对象的名称(app.py)，: 后面是任务函数名
'kwargs': {'job_name': 'job2'},
'trigger': 'cron',
'hour': 16, # 16 点执行
'minute': 58, # 58 分执行
'second': 3 # 3 秒执行
},
{
'id': 'job3',
'func': 'app:MyService.my_job', # 注意这里的格式，app 是 Flask 应用对象的名称(app.py)，: 后面是任务函数名
'kwargs': {'job_name': 'job3'},
'trigger': 'cron',
'hour': 16, # 16 点执行
'minute': 58, # 58 分执行
'second': 6 # 6 秒执行
}
]
# 开启API功能，这样才可以用api的方式去查看和修改定时任务
SCHEDULER_API_ENABLED = True
4.app.py中代码如下
from config.config import Config # 导入Config类的配置
from utils import init_scheduler # 导入init_scheduler方法
# 创建Flask应用
app = Flask(__name__)
app.config.from_object(Config) # 读取Config类的配置
init_scheduler(app) # init_scheduler方法5. MyService类中的my_job的方法使用app上下文
from flask import current_app # 导入flask的current_app(当前app)
from utils import scheduler # 很关键的一步 导入utils.__init__.py 初始化后的scheduler对象
class MyService:
@classmethod
def my_job(cls, job_name):
# # # 此方法在定时任务多的情况下，会有性能问题，少的情况没啥问题
# app = create_app()
# with app.app_context():
# current_app.logger.info("my_job已执行")
# # print(f"my_job,当前时间{datetime.now()}")
# # 使用全局APP变量
# get_app()
# with APP.app_context():
# current_app.logger.info(f"{job_name}已执行")
# print(f"my_job,当前时间{datetime.now()}")
with scheduler.app.app_context(): # 这个sheduler是带有app及其上下文的
current_app.logger.info(f"{job_name}已执行")不建议使用 创建一个app的方法
create app的链接：https://blog.csdn.net/weixin_41934979/article/details/140406152
6.执行效果如下： 
源代码地址：https://gitee.com/jxzcode_admin/flask-project.git
7.总结
使用的Python 软件包下的__init__.py文件中初始化生成scheduler对象,此对象项目启动后只生成一次，然后导入scheduler对象，在定时任务执行的方法使用: with scheduler.app.app_context(): 就可以 获取flask当前app上下文，不需要create app,个人觉得这才是真正正确使用Flask_APScheduler
参考资料
评论(0)