您的位置:首页 > 新闻 > 会展 > Python Flask_APScheduler定时任务的正确(最佳)使用

Python Flask_APScheduler定时任务的正确(最佳)使用

2024/12/24 11:16:04 来源:https://blog.csdn.net/weixin_41934979/article/details/141901997  浏览:    关键词:Python Flask_APScheduler定时任务的正确(最佳)使用

描述 

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_APScheduler

1.项目结构图如下:

 

 2.Python 软件包utils下的__init__.py 初始化生成APScheduler对象

这里可以灵活处理,例如:也可以是common软件包下__init__.py里初始化APScheduler

 __init__.py的代码如下:

import atexit
import platformfrom 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:passdef _unlock_file():try:f.seek(0)msvcrt.locking(f.fileno(), msvcrt.LK_UNLCK, 1)except Exception as e:passatexit.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:@classmethoddef 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

 参考资料

https://blog.csdn.net/arnolan/article/details/84936075

https://www.jianshu.com/p/d5a46b2d2fd3

版权声明:

本网仅为发布的内容提供存储空间,不对发表、转载的内容提供任何形式的保证。凡本网注明“来源:XXX网络”的作品,均转载自其它媒体,著作权归作者所有,商业转载请联系作者获得授权,非商业转载请注明出处。

我们尊重并感谢每一位作者,均已注明文章来源和作者。如因作品内容、版权或其它问题,请及时与我们联系,联系邮箱:809451989@qq.com,投稿邮箱:809451989@qq.com