对于有些flask系统,后台需要启动轮询线程,执行特定的任务,以下是一个简单的例子。
globals/daemon.py
import threading
from app.executor.ops_service import find_and_run_ops_task_todo_in_redisdef context_run_func(app, func):with app.app_context():func(app)def daemon_start(app):thread1 =threading.Thread(target=context_run_func, args=(app, find_and_run_ops_task_todo_in_redis))thread1.setDaemon(True)thread1.start()
init.py
from flask import Flask
from .globals.daemon import daemon_startflask_app = Flask(__name__)def create_app():CORS(flask_app, supports_credentials=True)with flask_app.app_context():config_init(flask_app, sys_args) # 初始化配置log_init(flask_app) # 初始化日志记录register_blueprint(flask_app) # 注册蓝图register_executor(flask_app) # 注册异步程序执行器register_scheduler(flask_app) # 注册计划任务daemon_start(flask_app)
run.py
myapp = create_app(sys_args)if __name__ == '__main__':myapp.run()
注意
- 不能以flask-executor启动异步执行的方式启动后台线程,会运行报错:executor需要request上下文;
- threading内需要传递app的上下文,否则会报错:未在app的上下文内。