Flask应用长任务导致504 Gateway Time-out的解决方法咨询

问题原因 #

504 Gateway Time-out是因为请求处理时间超过了网关（如Nginx）或Flask服务器的超时限制——虽然后台业务逻辑已执行完成，但网关已提前断开连接，导致无法返回结果。

临时解决方案：延长超时时间 #

1. Flask开发服务器（Werkzeug） #

如果使用Flask自带的开发服务器，启动时可设置timeout参数：

if __name__ == '__main__':
    app.run(timeout=120)  # 设置超时为120秒

注意：该方法仅适用于开发环境，生产环境不建议使用Werkzeug服务器。

2. 反向代理（如Nginx） #

若使用Nginx作为反向代理，需修改Nginx配置文件，调整超时参数：

location / {
    proxy_pass http://your_flask_server_address;
    proxy_connect_timeout 120s;  # 连接超时
    proxy_send_timeout 120s;     # 发送超时
    proxy_read_timeout 120s;     # 读取超时
}

修改后重启Nginx生效。

生产环境最优方案：异步任务处理 #

延长超时只是临时手段，长耗时任务更适合用异步架构处理，避免阻塞请求连接，提升系统稳定性和用户体验。

方案1：Flask异步视图（Flask 2.0+） #

如果任务是IO密集型，可直接使用Flask的异步支持：

import asyncio
from flask import Flask, request

app = Flask(__name__)

async def long_running_task(data):
    # 模拟IO密集型长耗时任务
    await asyncio.sleep(70)
    return f"处理结果: {data}"

@app.route('/v1/config', methods=['POST'])
async def post_income():
    content_type = request.headers.get('Content-Type')
    if content_type == 'application/json':
        json_in = request.get_json()
        value = await long_running_task(json_in)
        print(value)
        return value

若任务是CPU密集型，需结合线程池包装同步任务：

import concurrent.futures
from flask import Flask, request

app = Flask(__name__)

def long_running_task(data):
    # 模拟CPU密集型长耗时任务
    import time
    time.sleep(70)
    return f"处理结果: {data}"

@app.route('/v1/config', methods=['POST'])
async def post_income():
    content_type = request.headers.get('Content-Type')
    if content_type == 'application/json':
        json_in = request.get_json()
        with concurrent.futures.ThreadPoolExecutor() as executor:
            value = await app.loop.run_in_executor(executor, long_running_task, json_in)
        print(value)
        return value

方案2：任务队列（Celery + Redis/RabbitMQ） #

这是生产环境处理长耗时任务的标准方案，流程为：接口接收请求后返回任务ID，后台异步处理任务，用户通过任务ID查询结果。

步骤1：安装依赖

pip install celery redis

步骤2：定义Celery任务（tasks.py）

from celery import Celery

# 配置Redis作为消息代理和结果存储
app = Celery('tasks', broker='redis://localhost:6379/0', backend='redis://localhost:6379/0')

@app.task
def long_running_task(data):
    # 实际业务逻辑
    import time
    time.sleep(70)
    return f"处理结果: {data}"

步骤3：修改Flask接口

from flask import Flask, request, jsonify
from tasks import long_running_task

app = Flask(__name__)

@app.route('/v1/config', methods=['POST'])
def post_income():
    content_type = request.headers.get('Content-Type')
    if content_type == 'application/json':
        json_in = request.get_json()
        # 提交异步任务
        task = long_running_task.delay(json_in)
        return jsonify({"task_id": task.id, "状态": "任务已提交"})

@app.route('/v1/config/status/<task_id>', methods=['GET'])
def get_task_status(task_id):
    task = long_running_task.AsyncResult(task_id)
    if task.state == 'SUCCESS':
        return jsonify({"状态": task.state, "结果": task.result})
    else:
        return jsonify({"状态": task.state})

步骤4：启动Celery Worker

celery -A tasks worker --loglevel=info

总结 #

• 临时延长超时可快速解决问题，但不适用于高并发生产环境；
• 异步任务处理是更优方案，其中Celery任务队列适合复杂的生产场景，能有效避免超时问题，提升系统可扩展性。

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