对于有些flask系统，后台需要启动轮询线程，执行特定的任务，以下是一个简单的例子。文章来源地址https://www.toymoban.com/news/detail-660040.html
globals/daemon.py
import threading
from app.executor.ops_service import find_and_run_ops_task_todo_in_redis
def 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_start
flask_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的上下文内。
文章来源：https://www.toymoban.com/news/detail-660040.html
到了这里，关于Flask-flask系统运行后台轮询线程的文章就介绍完了。如果您还想了解更多内容，请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章，希望大家以后多多支持TOY模板网！
![[Python] flask运行+wsgi切换生产环境+supervisor配置指南](https://imgs.yssmx.com/Uploads/2024/02/774682-1.png)
