描述
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 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
参考资料
https://blog.csdn.net/arnolan/article/details/84936075
https://www.jianshu.com/p/d5a46b2d2fd3
免责声明:如果侵犯了您的权益,请联系站长,我们会及时删除侵权内容,谢谢合作!更多信息从访问主页:qidao123.com:ToB企服之家,中国第一个企服评测及商务社交产业平台。 |