最近在使用Python Flask项目开发的时候有个功能,我想使用多线程执行,执行过程中会操作数据库,开发好测试的时候报错:RuntimeError: No application found。
这个报错就是flask最常见的上下文问题,flask-sqlalchemy官方文档也给出了解决方案:https://flask-sqlalchemy.palletsprojects.com/en/2.x/contexts/
也就是手动推送上下文:
from atang.blog import apps as bp from atang import create_app from atang.extensions import scheduler app = create_app() @bp.route('/atang_blog') def AtangBlog(): # 推送上下文 with app.app_context(): print("当前计划任务状态:{}".format(scheduler.running)) url = app.config.get("ATANG_BLOG_URL") # print("阿汤博客:https://www.amd5.cn) print("阿汤博客:{}".format(url)) return {"name": "阿汤博客", "url": url}
但是这样手动推送上下文以后,出现了新的报错:
Traceback (most recent call last): File "F:\python\flask-test\flask_env\Lib\site-packages\flask\cli.py", line 68, in find_best_app app = call_factory(script_info, app_factory) File "F:\python\flask-test\flask_env\Lib\site-packages\flask\cli.py", line 123, in call_factory return app_factory(*args, **kwargs) File "F:\python\flask-test\ops\__init__.py", line 36, in create_app InitApp2(app) File "F:\python\flask-test\ops\extensions.py", line 17, in InitApp2 scheduler.init_app(app) File "F:\python\flask-test\flask_env\Lib\site-packages\flask_apscheduler\scheduler.py", line 83, in init_app self._load_config() File "F:\python\flask-test\flask_env\Lib\site-packages\flask_apscheduler\scheduler.py", line 316, in _load_config self._scheduler.configure(**options) File "F:\python\flask-test\flask_env\Lib\site-packages\apscheduler\schedulers\base.py", line 108, in configure raise SchedulerAlreadyRunningError apscheduler.schedulers.SchedulerAlreadyRunningError: Scheduler is already running
字面意思就是定时任务已经在运行。
和这样(flask run --host=0.0.0.0初始化时):
File "F:\python\flask-test\flask_env\lib\site-packages\flask_apscheduler\scheduler.py", line 83, in init_app self._load_config() File "F:\python\flask-test\flask_env\lib\site-packages\flask_apscheduler\scheduler.py", line 316, in _load_config self._scheduler.configure(**options) File "F:\python\flask-test\flask_env\lib\site-packages\apscheduler\schedulers\base.py", line 131, in configure self._configure(config) File "F:\python\flask-test\flask_env\lib\site-packages\apscheduler\schedulers\background.py", line 29, in _configure super(BackgroundScheduler, self)._configure(config) File "F:\python\flask-test\flask_env\lib\site-packages\apscheduler\schedulers\base.py", line 727, in _configure raise ValueError( ValueError: Cannot create executor "default" -- either "type" or "class" must be defined
字面意思就是没有定义默认参数。
很遗憾找了好几个小时,没找到解决方案。
加了两个flask的开发群想请教咨询下大佬,结果发了以后没人回,全在灌水,没办法只能自己想办法。
经过测试只要注释掉Flask-APScheduler的初始化代码scheduler.init_app(app)和scheduler.start()上下文推送就正常,多线程运行也正常。
经过几个小时的踩坑我都想放弃使用多线程了,只是效率低一点还能接受,因为有一些定时任务,不能放弃Flask-APScheduler不用。但是后面我在看Flask-APScheduler的报错相关的源码并打印相关参数时,发现Flask-APScheduler重复初始化了,相关报错源码:
当我运行的flask run --host=0.0.0.0时,他会重复加载两次:
网上找了下原因:
当调用app.run()的时候,用到了Werkzeug库,它会生成一个子进程,当代码有变动的时候它会自动重启。
如果在run()里加入参数 use_reloader=False,就会取消这个功能,当然代码改动后也不会自动更新了。
当然这个加载两次,在非debug模式不会出现。
debug模式或者开发环境可以通过判断是不是werkzeug线程选择加载,这个我也是在看Flask-APScheduler官方例子的时候发现的解决方案。
当我手动推送上下文调用:
from ops import create_app app = create_app()
相当于又要重新初始化一次,所以Flask-APScheduler抛出了异常:apscheduler.schedulers.SchedulerAlreadyRunningError: Scheduler is already running。
那我想办法在手动推送上下文的时候不执行Flask-APScheduler的初始化代码,不就正常了吗。
带着这个想法,又去网上找了找解决方案,这次方向总算对了,有点眉目了,网上找到了Flask-APScheduler重复执行任务的解决办法,那就是使用一个全局锁。
网上还有一种方案就把Flask-APScheduler的初始化代码放在if __name__ == '__main__':后面,这种方案灵活性太低了。
这里说说全局锁的原理:应用启动,第一次初始化Flask-APScheduler的时候,打开一个文件,然后给这个文件加一个非阻塞排他锁;如果加锁失败,说明Flask-APScheduler已经初始化了,就略过。
方案使用的是fcntl文件锁,但是fcntl只能在Unix平台运行,Windows平台不兼容。
因为我开发用的电脑是Windows,只能又找了一个跨平台的文件锁模块portalocker,看了portalocker的源码,他的实现也是使用的fcntl(Unix)和win32 api(Windows)。
修改以后的初始化代码:
from flask_apscheduler import APScheduler import portalocker import atexit scheduler = APScheduler() def InitApp2(app): file = open("scheduler.lock", "wb") try: # 加排他非阻塞锁,LOCK_EX 排他锁 LOCK_NB 非阻塞锁 portalocker.lock(file, portalocker.LOCK_EX|portalocker.LOCK_NB) print("文件上锁成功!") scheduler.init_app(app) scheduler.start() except Exception as e: print("文件已锁:{}".format(e)) pass def Unlock(): # 解锁 portalocker.unlock(file) file.close() # 将 func注册为终止时执行的函数. atexit.register(Unlock)
代码重新加载以后,测试一切都正常。
但是当我停止应用,再执行flask run --host=0.0.0.0以后,计划任务确没有运行。
通过分析就是因为上面我提到的Werkzeug导致的初始化两次。
而加锁是在第一次非Werkzeug子进程的时候(看上面的图,是在Debug mode: on后面),导致后面初始化时候,文件已经加锁,导致初始化失败。
我前面提到过解决方案,就是判断下是不是Werkzeug的子进程。所以修改以后的代码如下:
from flask_apscheduler import APScheduler import portalocker import atexit impot os scheduler = APScheduler() def InitApp2(app): # 开发环境 file = open("scheduler.lock", "wb") # 上锁 def Lock(): try: # 加排他非阻塞锁, LOCK_EX 排他锁 、LOCK_NB 非阻塞锁 portalocker.lock(file, portalocker.LOCK_EX | portalocker.LOCK_NB) # 初始化Flask-APScheduler,定时任务 scheduler.init_app(app) scheduler.start() except: pass # 非开发环境直接上锁 if os.environ.get("FLASK_ENV") == "development": # 如果非WERKZEUG 子进程跳过,防止debug模式提前加锁 if os.environ.get("WERKZEUG_RUN_MAIN"): Lock() else: Lock() # 解锁 def Unlock(): # 解锁 portalocker.unlock(file) file.close() # 将 func 注册为终止时执行的函数. atexit.register(Unlock)
再次执行flask run --host=0.0.0.0后一切正常(加锁到了Debugger PIN后面):