Flask中使用APScheduler并发执行任务时遭遇应用上下文错误
首先得帮你搞明白为啥会报这个错:你用BackgroundScheduler跑的定时任务,是在Flask的请求线程之外独立运行的,而Flask的**应用上下文(app_context)**是个包含应用核心配置、全局状态的环境——像jsonify这种函数,必须在这个环境里才能工作,因为它要读取app的JSON相关配置来生成响应。你的定时任务没带上这个上下文,自然就触发了Working outside of application context的错误。
接下来给你两种直接能用的解决办法,适配你现在的代码:
方法一:在chron函数内部手动维护上下文
最简单的方式,就是在chron里需要调用Flask相关函数(比如jsonify)的代码块外,用with语句包裹应用上下文。这样每次任务执行时,都会自动推入上下文,执行完再弹出,不会有内存泄漏问题:
from flask import jsonify, current_app def chron(obj): # 用with语句手动创建应用上下文 with current_app.app_context(): # 这里写你原来的业务逻辑,比如调用jsonify response_data = {"task_id": obj, "status": "completed"} json_response = jsonify(response_data) # 其他需要上下文的操作也放这里面 print(f"任务 {obj} 执行成功,响应: {json_response.get_data(as_text=True)}")
注意:如果你的Flask app实例是在单独的模块里定义的(比如app = Flask(__name__)写在app.py里),也可以直接导入app,用with app.app_context():,效果是一样的。
方法二:添加任务时预先绑定上下文
如果你的定时任务很多,不想每个任务函数都写一遍上下文代码,可以在etl函数添加任务的时候,把chron包装在上下文里:
def etl(app): # 把你的Flask app实例传进来 with open('appConfig.json', encoding="utf-8") as json_file: configData = json.load(json_file) for obj in configData: # 定义一个包装函数,自动带上应用上下文 def wrapped_chron(): with app.app_context(): chron(obj) # 添加包装后的函数作为定时任务 sched.add_job(wrapped_chron, 'interval', seconds=5, id=obj)
这样每个任务执行时,都会自动进入应用上下文,chron里的代码就能正常调用jsonify了。
额外提一句:如果以后你需要更省心的定时任务集成,可以试试Flask-APScheduler扩展——它是APScheduler针对Flask的封装,已经自动处理了应用上下文的问题,不用你手动写这些代码。不过就你现在的场景,上面两种方法完全能解决问题。
内容的提问来源于stack exchange,提问作者Mithil Bhoras

