Flask-flask系统运行后台轮询线程
对于有些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的上下文内。