Flask应用中长耗时进程如何将向浏览器发送的请求数限制为1？

核心思路 #

• 解耦HTTP请求生命周期与长耗时任务：将深度学习文件处理逻辑从Flask主进程剥离，交由异步任务队列托管，请求提交后立即返回任务ID，不会阻塞等待30分钟，从根源避免浏览器超时重发
• 任务去重：相同文件/参数的未完成任务不会重复提交
• 前端轮询查状态：前端拿到任务ID后定期主动查询进度，无需保持长连接

依赖安装 #

执行命令安装所需组件：
pip install flask celery redis
使用Celery作为异步任务队列，Redis作为消息中间件和任务状态存储，是Flask生态最成熟的长任务解决方案。

完整代码实现 #

1. 基础配置与初始化 #

from flask import Flask, request, jsonify
from celery import Celery
import hashlib
import os

app = Flask(__name__)
# Celery配置，Redis默认端口6379
app.config['CELERY_BROKER_URL'] = 'redis://localhost:6379/0'
app.config['CELERY_RESULT_BACKEND'] = 'redis://localhost:6379/0'

# 绑定Celery与Flask上下文
def make_celery(app):
    celery = Celery(
        app.import_name,
        broker=app.config['CELERY_BROKER_URL'],
        backend=app.config['CELERY_RESULT_BACKEND']
    )
    celery.conf.update(app.config)
    class ContextTask(celery.Task):
        def __call__(self, *args, **kwargs):
            with app.app_context():
                return self.run(*args, **kwargs)
    celery.Task = ContextTask
    return celery

celery = make_celery(app)
# 任务去重缓存，生产环境建议存Redis避免Flask重启丢失
task_cache = {}

2. 定义长耗时处理任务 #

@celery.task(bind=True, time_limit=3600) # 设置1小时超时避免资源泄漏
def process_large_file_task(self, file_path, output_dir):
    # 替换为你的深度学习文件分割逻辑
    total_step = 100 # 按实际处理流程拆分步骤，用于计算进度
    for step in range(total_step):
        # 模拟30分钟处理时长
        import time
        time.sleep(18)
        # 更新任务进度，供前端查询
        self.update_state(state='PROGRESS', meta={'current': step+1, 'total': total_step})
    
    # 处理完成后返回结果
    output_file = os.path.join(output_dir, f"processed_{os.path.basename(file_path)}.zip")
    return {"download_url": f"/download/{os.path.basename(output_file)}"}

3. 接口实现（含去重逻辑） #

@app.route('/submit-task', methods=['POST'])
def submit_task():
    # 计算上传文件哈希作为去重唯一标识
    uploaded_file = request.files['file']
    file_content = uploaded_file.read()
    file_hash = hashlib.md5(file_content).hexdigest()
    uploaded_file.seek(0) # 重置文件指针用于后续存储

    # 去重校验：相同文件的运行中任务直接返回已有ID
    if file_hash in task_cache:
        exist_task = process_large_file_task.AsyncResult(task_cache[file_hash])
        if exist_task.state in ['PENDING', 'PROGRESS']:
            return jsonify({
                "task_id": task_cache[file_hash],
                "msg": "该文件已在处理中"
            })
        else:
            # 已结束的任务清除缓存
            del task_cache[file_hash]
    
    # 存储上传文件，不要直接传递文件内容到Celery
    save_dir = "./upload"
    os.makedirs(save_dir, exist_ok=True)
    save_path = os.path.join(save_dir, f"{file_hash}_{uploaded_file.filename}")
    uploaded_file.save(save_path)

    # 提交异步任务
    output_dir = "./output"
    os.makedirs(output_dir, exist_ok=True)
    task = process_large_file_task.delay(save_path, output_dir)
    task_cache[file_hash] = task.id

    return jsonify({
        "task_id": task.id,
        "msg": "任务提交成功"
    })

@app.route('/task-status/<task_id>', methods=['GET'])
def get_task_status(task_id):
    task = process_large_file_task.AsyncResult(task_id)
    if task.state == 'PENDING':
        return jsonify({"state": "pending", "progress": 0})
    elif task.state == 'PROGRESS':
        progress = int(task.info['current'] / task.info['total'] * 100)
        return jsonify({"state": "progress", "progress": progress})
    elif task.state == 'SUCCESS':
        return jsonify({"state": "success", "result": task.result})
    else:
        return jsonify({"state": "failed", "error": str(task.info)})

服务启动步骤 #

• 启动本地Redis服务（默认端口6379）
• 启动Celery worker节点：celery -A app.celery worker --loglevel=info
• 启动Flask主服务：flask run

前端侧配套修改 #

• 提交任务后拿到task_id立即结束请求，不要设置长超时时间，避免触发浏览器超时重传机制
• 用定时器每5~10秒调用/task-status/<task_id>接口查询进度，实时展示给用户
• 可将task_id存在本地存储，页面刷新后可直接查询对应任务进度，无需重复上传

额外优化建议 #

• 多用户场景下可在去重标识中加入用户ID，避免不同用户的相同文件互相影响
• 任务缓存改用Redis存储，支持多实例部署且避免Flask重启后缓存丢失
• 增加任务结果过期清理逻辑，避免存储冗余文件

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