基于Python与Flask：如何在用户等待时保持连接存活避免超时？

针对你遇到的大文件生成耗时过长导致连接超时的问题，这里提供几种实用的解决方案，按落地难度和适用场景排序：

1. 分块流式输出（最快落地） #

核心思路是先发送响应头，然后在文件生成过程中定期输出空字节或内容块，保持HTTP连接活跃，避免客户端判定超时。Flask的Response支持接收生成器作为数据源，完美适配这种场景。

示例代码：

from flask import Flask, Response
import time

app = Flask(__name__)

def generate_large_file():
    # 先发送空字节触发响应头返回，客户端会保持连接
    yield b''
    # 模拟大文件生成逻辑（替换成你的实际代码）
    total_chunks = 100
    for chunk_idx in range(total_chunks):
        # 生成当前块内容
        content_chunk = f"File content part {chunk_idx + 1}/{total_chunks}\n".encode('utf-8')
        yield content_chunk
        # 模拟每块生成耗时2秒
        time.sleep(2)
        # 每间隔一段时间发送空字节，防止连接被断开
        yield b''

@app.route('/download-large-file')
def download():
    return Response(
        generate_large_file(),
        headers={
            'Content-Disposition': 'attachment; filename="generated_file.txt"',
            'Content-Type': 'text/plain',
            # 明确告知客户端这是分块响应
            'Transfer-Encoding': 'chunked'
        }
    )

if __name__ == '__main__':
    app.run()

这种方式无需依赖额外服务，直接修改现有接口即可，适合大部分场景。

2. 异步任务+轮询/SSE（适合超长时间任务） #

如果文件生成耗时经常超过5分钟，建议把生成逻辑放到后台异步任务中，客户端通过轮询或Server-Sent Events（SSE）获取进度，完成后再下载文件。

实现步骤： #

1. 用Celery+Redis（或RabbitMQ）搭建异步任务队列
2. 客户端请求触发任务，获取任务ID
3. 客户端轮询任务状态，或通过SSE实时接收进度
4. 任务完成后，客户端请求下载生成好的文件

示例代码（Celery+Flask）：

异步任务配置

# celery_tasks.py
from celery import Celery
import time
import os

celery = Celery(
    'file_generator',
    broker='redis://localhost:6379/0',
    backend='redis://localhost:6379/0'
)

@celery.task(bind=True)
def generate_large_file_task(self):
    # 更新任务进度（可选）
    self.update_state(state='PROGRESS', meta={'current': 0, 'total': 100})
    # 模拟大文件生成
    temp_file_path = '/tmp/generated_large_file.txt'
    with open(temp_file_path, 'w') as f:
        for i in range(100):
            f.write(f"Line {i+1}\n")
            time.sleep(1.2)
            self.update_state(state='PROGRESS', meta={'current': i+1, 'total': 100})
    return temp_file_path

Flask接口

from flask import Flask, jsonify, send_file
from celery_tasks import generate_large_file_task
import os

app = Flask(__name__)

@app.route('/start-generation')
def start_generation():
    task = generate_large_file_task.delay()
    return jsonify({'task_id': task.id})

@app.route('/check-task/<task_id>')
def check_task_status(task_id):
    task = generate_large_file_task.AsyncResult(task_id)
    if task.state == 'SUCCESS':
        return jsonify({
            'state': 'completed',
            'download_url': f'/download/{task_id}'
        })
    elif task.state == 'PROGRESS':
        return jsonify({
            'state': 'in_progress',
            'progress': task.info
        })
    return jsonify({'state': task.state})

@app.route('/download/<task_id>')
def download_generated_file(task_id):
    task = generate_large_file_task.AsyncResult(task_id)
    if task.state != 'SUCCESS':
        return jsonify({'error': 'File not ready yet'}), 400
    file_path = task.result
    if not os.path.exists(file_path):
        return jsonify({'error': 'File not found'}), 404
    response = send_file(file_path, as_attachment=True, download_name='large_file.txt')
    # 下载完成后删除临时文件（可选）
    os.remove(file_path)
    return response

if __name__ == '__main__':
    app.run()

这种方式彻底解决长连接占用问题，还能给用户展示进度，提升体验。

3. 调整超时配置（辅助手段） #

即使使用上面的方案，也需要同步调整相关组件的超时设置：

• WSGI服务器：比如Gunicorn默认超时30秒，启动时通过--timeout 300（5分钟）调整；uWSGI通过harakiri = 300设置。
• 反向代理：如果用Nginx做反向代理，需要调整proxy_connect_timeout、proxy_send_timeout、proxy_read_timeout参数，设置为大于文件生成的最大耗时。
• 客户端：如果是自定义客户端，需要将请求超时时间设置为足够大的值；浏览器端通常默认超时时间较长，但仍需注意。

注意事项 #

• 流式输出时，确保客户端支持分块响应（主流浏览器和HTTP客户端都支持）。
• 异步任务方案中，要做好临时文件的清理，避免磁盘空间被占满。
• 超长时间任务建议添加任务超时和失败重试机制。

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