如何维持Server-Sent Events连接？Quart后端连接频繁中断问题

使用Quart框架搭建作为Event Source的后端API，负责处理一项耗时约2分钟的任务。为维持连接，计划每10秒发送一次心跳包，React前端能够正常接收事件，但存在的问题是每过1分钟API请求就会被取消并重新发起同一接口的请求。希望在耗时任务完成前持续向客户端发送心跳包，保持连接不中断。

后端Python（Quart）代码 #

import asyncio
from quart import Quart, Response, websocket

app = Quart(__name__)

@app.after_request
async def add_cors_headers(response):
    response.headers['Access-Control-Allow-Origin'] = 'http://localhost:3000'
    response.headers['Access-Control-Allow-Methods'] = 'GET, POST, PUT, DELETE'
    response.headers['Access-Control-Allow-Headers'] = 'Content-Type'
    return response

async def time_consuming_task(queue):
    # 模拟耗时任务
    await asyncio.sleep(120)
    # 任务完成后将结果放入队列
    await queue.put("Task completed")

async def in_progress_events(queue):
    i = 0
    while True:
        yield f"data: {i}\n\n"
        i += 1
        await asyncio.sleep(1)

@app.route('/sse')
async def sse():
    # 创建队列用于和耗时任务通信
    queue = asyncio.Queue()

    # 启动耗时任务
    asyncio.create_task(time_consuming_task(queue))

    async def event_stream():
        while True:
            async for event in in_progress_events(queue):
                yield event

                # 检查队列中是否有结果
                if not queue.empty():
                    result = await queue.get()
                    yield f"data: {result}\n\n"
                    yield "event: close\ndata: Connection closed\n\n"
                    return

            # 发送心跳事件保持连接
            yield "event: heartbeat\ndata: \n\n"
            await asyncio.sleep(50)  # 每50秒发送一次心跳

    response = Response(event_stream(), content_type='text/event-stream')
    response.headers['Connection'] = 'keep-alive'
    response.headers['Cache-Control'] = 'no-cache'
    response.headers['Transfer-Encoding'] = 'chunked'
    return response

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

前端React代码 #

import React, { useEffect, useState } from 'react';

function SSEComponent() {
  const [data, setData] = useState('');

  useEffect(() => {
    const eventSource = new EventSource('http://localhost:5000/sse');

    eventSource.onmessage = (event) => {
        console.log("onmessage", event.data);
      setData(event.data);
    };

    eventSource.addEventListener('close', () => {
        console.log("onclose");
      eventSource.close();
    });

    eventSource.addEventListener('heartbeat', () => {
        // 处理心跳事件
        console.log('Received heartbeat event');
      });

    eventSource.onerror = (error) => {
        console.log("error", error);
    };

    return () => {
      eventSource.close();
    };
  }, []);

  return (
    <div>
      <h2>Server-Sent Events Data:</h2>
      <p>{data}</p>
    </div>
  );
}

export default SSEComponent;

问题原因 #

1. 后端心跳逻辑失效：当前代码中event_stream函数里，async for event in in_progress_events(queue)是无限循环（in_progress_events内为while True），导致下方的心跳发送代码永远不会执行，等于没有实际发送心跳。
2. 连接超时触发重连：由于没有有效心跳数据，浏览器或中间件（如Nginx）会在1分钟左右判定连接超时，主动断开并重新发起请求。

解决方案 #

调整后端事件生成逻辑，将进度事件与心跳事件合并到同一循环中，确保持续发送数据维持连接：

修改后的后端代码 #

import asyncio
from quart import Quart, Response

app = Quart(__name__)

@app.after_request
async def add_cors_headers(response):
    response.headers['Access-Control-Allow-Origin'] = 'http://localhost:3000'
    response.headers['Access-Control-Allow-Methods'] = 'GET, POST, PUT, DELETE'
    response.headers['Access-Control-Allow-Headers'] = 'Content-Type'
    return response

async def time_consuming_task(queue):
    # 模拟耗时任务
    await asyncio.sleep(120)
    # 任务完成后将结果放入队列
    await queue.put("Task completed")

@app.route('/sse')
async def sse():
    queue = asyncio.Queue()
    asyncio.create_task(time_consuming_task(queue))

    async def event_stream():
        i = 0
        while True:
            # 先检查任务是否完成
            if not queue.empty():
                result = await queue.get()
                yield f"data: {result}\n\n"
                yield "event: close\ndata: Connection closed\n\n"
                return
            
            # 发送进度事件
            yield f"data: {i}\n\n"
            i += 1
            
            # 每10秒发送一次心跳事件
            if i % 10 == 0:
                yield "event: heartbeat\ndata: \n\n"
            
            await asyncio.sleep(1)

    response = Response(event_stream(), content_type='text/event-stream')
    response.headers['Connection'] = 'keep-alive'
    response.headers['Cache-Control'] = 'no-cache'
    response.headers['Transfer-Encoding'] = 'chunked'
    # 禁用反向代理缓冲，确保实时推送
    response.headers['X-Accel-Buffering'] = 'no'
    # 设置长连接超时为5分钟，覆盖默认短超时
    response.headers['Keep-Alive'] = 'timeout=300, max=1000'
    return response

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

关键调整说明 #

• 将进度事件与心跳事件合并到同一循环，确保每1秒都有数据发送，远低于1分钟超时阈值，避免连接被断开。
• 添加X-Accel-Buffering: no头部，防止Nginx等反向代理缓冲响应导致客户端无法实时接收数据。
• 设置Keep-Alive头部明确指定连接超时时间为5分钟，覆盖默认的短超时配置。

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