最近在使用Python Flask項(xiàng)目開發(fā)的時(shí)候有個(gè)功能，我想使用多線程執(zhí)行，執(zhí)行過程中會(huì)操作數(shù)據(jù)庫(kù)，開發(fā)好測(cè)試的時(shí)候報(bào)錯(cuò)：RuntimeError: No application found。
這個(gè)報(bào)錯(cuò)就是flask最常見的上下文問題，flask-sqlalchemy官方文檔也給出了解決方案：https://flask-sqlalchemy.palletsprojects.com/en/2.x/contexts/
也就是手動(dòng)推送上下文：
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("當(dāng)前計(jì)劃任務(wù)狀態(tài)：{}".format(scheduler.running))
url = app.config.get("ATANG_BLOG_URL")
# print("阿湯博客：http://www.zhongjima.net)
print("阿湯博客：{}".format(url))
return {"name": "阿湯博客", "url": url}
但是這樣手動(dòng)推送上下文以后，出現(xiàn)了新的報(bào)錯(cuò)：
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
字面意思就是定時(shí)任務(wù)已經(jīng)在運(yùn)行。
和這樣（flask run --host=0.0.0.0初始化時(shí)）：
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
字面意思就是沒有定義默認(rèn)參數(shù)。
很遺憾找了好幾個(gè)小時(shí)，沒找到解決方案。
加了兩個(gè)flask的開發(fā)群想請(qǐng)教咨詢下大佬，結(jié)果發(fā)了以后沒人回，全在灌水，沒辦法只能自己想辦法。
經(jīng)過測(cè)試只要注釋掉Flask-APScheduler的初始化代碼scheduler.init_app(app)和scheduler.start()上下文推送就正常，多線程運(yùn)行也正常。
經(jīng)過幾個(gè)小時(shí)的踩坑我都想放棄使用多線程了，只是效率低一點(diǎn)還能接受，因?yàn)橛幸恍┒〞r(shí)任務(wù)，不能放棄Flask-APScheduler不用。但是后面我在看Flask-APScheduler的報(bào)錯(cuò)相關(guān)的源碼并打印相關(guān)參數(shù)時(shí)，發(fā)現(xiàn)Flask-APScheduler重復(fù)初始化了，相關(guān)報(bào)錯(cuò)源碼：
當(dāng)我運(yùn)行的flask run --host=0.0.0.0時(shí)，他會(huì)重復(fù)加載兩次：
網(wǎng)上找了下原因：
當(dāng)調(diào)用app.run()的時(shí)候，用到了Werkzeug庫(kù)，它會(huì)生成一個(gè)子進(jìn)程，當(dāng)代碼有變動(dòng)的時(shí)候它會(huì)自動(dòng)重啟。
如果在run()里加入?yún)?shù) use_reloader=False，就會(huì)取消這個(gè)功能，當(dāng)然代碼改動(dòng)后也不會(huì)自動(dòng)更新了。
當(dāng)然這個(gè)加載兩次，在非debug模式不會(huì)出現(xiàn)。
debug模式或者開發(fā)環(huán)境可以通過判斷是不是werkzeug線程選擇加載，這個(gè)我也是在看Flask-APScheduler官方例子的時(shí)候發(fā)現(xiàn)的解決方案。
當(dāng)我手動(dòng)推送上下文調(diào)用：
from ops import create_app app = create_app()
相當(dāng)于又要重新初始化一次，所以Flask-APScheduler拋出了異常：apscheduler.schedulers.SchedulerAlreadyRunningError: Scheduler is already running。
那我想辦法在手動(dòng)推送上下文的時(shí)候不執(zhí)行Flask-APScheduler的初始化代碼，不就正常了嗎。
帶著這個(gè)想法，又去網(wǎng)上找了找解決方案，這次方向總算對(duì)了，有點(diǎn)眉目了，網(wǎng)上找到了Flask-APScheduler重復(fù)執(zhí)行任務(wù)的解決辦法，那就是使用一個(gè)全局鎖。
網(wǎng)上還有一種方案就把Flask-APScheduler的初始化代碼放在if __name__ == '__main__':后面，這種方案靈活性太低了。
這里說說全局鎖的原理：應(yīng)用啟動(dòng)，第一次初始化Flask-APScheduler的時(shí)候，打開一個(gè)文件，然后給這個(gè)文件加一個(gè)非阻塞排他鎖；如果加鎖失敗，說明Flask-APScheduler已經(jīng)初始化了，就略過。
方案使用的是fcntl文件鎖，但是fcntl只能在Unix平臺(tái)運(yùn)行，Windows平臺(tái)不兼容。
因?yàn)槲议_發(fā)用的電腦是Windows，只能又找了一個(gè)跨平臺(tái)的文件鎖模塊portalocker，看了portalocker的源碼，他的實(shí)現(xiàn)也是使用的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注冊(cè)為終止時(shí)執(zhí)行的函數(shù).
atexit.register(Unlock)
代碼重新加載以后，測(cè)試一切都正常。
但是當(dāng)我停止應(yīng)用，再執(zhí)行flask run --host=0.0.0.0以后，計(jì)劃任務(wù)確沒有運(yùn)行。
通過分析就是因?yàn)樯厦嫖姨岬降腤erkzeug導(dǎo)致的初始化兩次。
而加鎖是在第一次非Werkzeug子進(jìn)程的時(shí)候（看上面的圖，是在Debug mode: on后面），導(dǎo)致后面初始化時(shí)候，文件已經(jīng)加鎖，導(dǎo)致初始化失敗。
我前面提到過解決方案，就是判斷下是不是Werkzeug的子進(jìn)程。所以修改以后的代碼如下：
from flask_apscheduler import APScheduler
import portalocker
import atexit
impot os
scheduler = APScheduler()
def InitApp2(app):
# 開發(fā)環(huán)境
file = open("scheduler.lock", "wb")
# 上鎖
def Lock():
try:
# 加排他非阻塞鎖, LOCK_EX 排他鎖 、LOCK_NB 非阻塞鎖
portalocker.lock(file, portalocker.LOCK_EX | portalocker.LOCK_NB)
# 初始化Flask-APScheduler，定時(shí)任務(wù)
scheduler.init_app(app)
scheduler.start()
except:
pass
# 非開發(fā)環(huán)境直接上鎖
if os.environ.get("FLASK_ENV") == "development":
# 如果非WERKZEUG 子進(jìn)程跳過，防止debug模式提前加鎖
if os.environ.get("WERKZEUG_RUN_MAIN"):
Lock()
else:
Lock()
# 解鎖
def Unlock():
# 解鎖
portalocker.unlock(file)
file.close()
# 將 func 注冊(cè)為終止時(shí)執(zhí)行的函數(shù).
atexit.register(Unlock)
再次執(zhí)行flask run --host=0.0.0.0后一切正常（加鎖到了Debugger PIN后面）：
