Flask服务器能否实现多线程？Flask-RESTX多线程方案咨询

问题核心 #

你需要在Flask-RESTX服务中实现多线程运行耗时函数（比如从外部服务获取字符串的get_answer(a,b)），同时保证客户端能正常访问接口，当特定条件满足时，接口能正确返回符合parent模型的JSON数据。当前代码的问题在于模型定义阶段就直接调用了get_answer，而非请求时动态执行，导致数据无法按需更新。

先修正代码错误 #

你当前的模型定义存在逻辑问题：

parent = api.model('Parent', {
    'name': fields.String(get_answer(a,b)),  # 错误：模型初始化时就执行了get_answer，不是请求时动态获取
    'class': fields.String(discriminator=True)
})

这种写法会在Flask启动时就调用get_answer，而非客户端发起请求时执行，不仅无法动态获取最新数据，若get_answer是耗时操作，还会阻塞服务启动。

多线程实现方案 #

方案1：请求内开启线程处理耗时任务 #

如果需要客户端等待任务完成后返回结果，或者后台处理任务同时返回基础响应，可以在接口的get方法内手动开启线程：

from flask_restx import Api, fields, Resource
from flask import Flask
import threading

app = Flask(__name__)
api = Api(app)

# 先定义模型，字段仅做结构声明，不直接调用耗时函数
parent = api.model('Parent', {
    'name': fields.String(description='从外部服务获取的字符串'),
    'class': fields.String(discriminator=True)
})

def get_answer(a, b):
    # 模拟外部服务调用的耗时操作
    import time
    time.sleep(3)
    return f"result_{a}_{b}"

@api.route('/language')
class Language(Resource):
    @api.marshal_with(parent)
    @api.response(403, "Unauthorized")
    def get(self):
        # 解析请求参数，判断特定条件
        args = api.parser().add_argument('flag', type=bool, required=False).parse_args()
        response_data = {"class": "default_class"}
        
        if args.get('flag'):
            # 开启线程处理耗时任务
            def thread_task():
                nonlocal response_data
                response_data['name'] = get_answer(1, 2)
            
            thread = threading.Thread(target=thread_task)
            thread.start()
            # 如果需要等待任务完成再返回结果，取消下一行注释；否则直接返回基础数据，后续可通过轮询获取结果
            thread.join()
        
        return response_data

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=8080, threaded=True)  # Flask默认开启多线程处理请求

注意：Flask的app.run()默认threaded=True，会为每个请求分配独立线程，但请求内的额外耗时任务需要手动开启线程处理。

方案2：使用Celery异步任务 #

Celery完全兼容Flask-RESTX，适合处理不需要客户端同步等待的后台任务，比如异步获取外部数据后，客户端通过任务ID查询结果：

1. 先安装依赖：pip install celery redis（用Redis作为消息队列）
2. 集成代码示例：

from flask_restx import Api, fields, Resource
from flask import Flask
from celery import Celery

app = Flask(__name__)
# 配置Celery
app.config['CELERY_BROKER_URL'] = 'redis://localhost:6379/0'
app.config['CELERY_RESULT_BACKEND'] = 'redis://localhost:6379/0'

celery = Celery(app.name, broker=app.config['CELERY_BROKER_URL'])
celery.conf.update(app.config)

api = Api(app)

parent = api.model('Parent', {
    'name': fields.String(description='异步任务结果'),
    'class': fields.String(discriminator=True),
    'task_id': fields.String(description='异步任务ID，用于查询结果')
})

@celery.task
def get_answer_task(a, b):
    import time
    time.sleep(3)
    return f"result_{a}_{b}"

@api.route('/language')
class Language(Resource):
    @api.marshal_with(parent)
    @api.response(403, "Unauthorized")
    def get(self):
        args = api.parser().add_argument('flag', type=bool, required=False).parse_args()
        response_data = {"class": "default_class"}
        
        if args.get('flag'):
            # 启动异步任务
            task = get_answer_task.delay(1, 2)
            response_data['task_id'] = task.id
        
        return response_data

# 添加任务结果查询接口
@api.route('/task/<string:task_id>')
class TaskResult(Resource):
    def get(self, task_id):
        task = get_answer_task.AsyncResult(task_id)
        if task.state == 'SUCCESS':
            return {"name": task.result, "class": "default_class"}
        else:
            return {"status": task.state}, 202

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=8080)

使用时需要单独启动Celery worker：celery -A 你的脚本文件名 worker --loglevel=info，同时确保Redis服务正常运行。

你的尝试点评 #

把装饰器和Resource类封装到函数并开线程的方法是错误的，因为Flask-RESTX的路由和模型与应用实例绑定，动态创建路由会导致注册混乱，且无法保证线程安全。正确的做法是在请求处理逻辑（即Resource的方法内部）处理多线程/异步任务，而非修改路由注册逻辑。

内容的提问来源于stack exchange，提问作者Youkijiudai