当前位置：首页 > 文章列表 > 文章 > python教程 > Flask实现SSE实时推送方法详解

Flask实现SSE实时推送方法详解

2026-03-16 21:42:50 0浏览收藏

Flask实现SSE（服务端事件）看似只需几行yield代码，实则是一场贯穿前后端与部署全链路的精密协作——从后端必须用生成器返回符合data: ...\n\n格式的流式响应、正确设置Content-Type和no-cache头，到前端用EventSource手动实现健壮的断线重连逻辑，再到生产环境强制切换为uvicorn或gunicorn+gevent等异步服务器以避免同步模型下的连接阻塞与崩溃；任何一环出错（如少一个换行、漏配CORS、未禁用缓存、或在生成器中执行耗时操作）都会导致“页面没反应”“收不到消息”“CPU飙高”等典型故障，而这些问题往往藏在浏览器控制台看不见的底层连接细节里，真正考验的是对HTTP长连接本质的理解与系统性排错能力。

Flask Server-Sent Events怎么写_SSE向浏览器主动推送实时数据

Flask 怎么写 SSE 路由

Flask 本身不内置 SSE 支持，但靠 yield + 正确的响应头就能实现。核心是返回一个生成器（generator），每条数据以 data: ...\n\n 格式输出，并设好 Content-Type: text/event-stream 和禁用缓存。

常见错误是直接 return 字符串、忘了加双换行、或没设 Cache-Control: no-cache —— 浏览器会等缓冲填满才吐数据，导致“一直没反应”。

路由必须用 @app.route，不能用 @app.get（老版本 Flask 不识别）
函数返回值必须是生成器函数（含 yield），不能是普通字符串或列表
每条消息末尾必须有两个换行：yield "data: hello\n\n"，少一个就解析失败
推荐加 response.headers["Cache-Control"] = "no-cache"，避免 Nginx 或浏览器缓存住流

from flask import Flask, Response
import time

app = Flask(__name__)

@app.route('/sse')
def sse():
    def event_stream():
        for i in range(5):
            yield f"data: message {i}\n\n"
            time.sleep(1)
    return Response(event_stream(), content_type='text/event-stream')

前端怎么连上 Flask SSE 并处理断线重连

EventSource 是浏览器原生 API，简单可靠，但默认不自动重连 —— 断网、服务重启后它就停了，用户得刷新页面。这不是 bug，是规范行为。

容易踩的坑：没监听 error 事件、把 onmessage 写成 addEventListener('message', ...)（漏掉 on 前缀）、或在 onerror 里直接 new EventSource(...) 导致无限新建连接。

必须监听 onerror，并在其中检查 event.target.readyState 是否为 0（关闭状态）再重建
重连前建议加延迟（如 setTimeout），避免雪崩请求
如果后端返回自定义事件类型（如 event: update\n），前端要用 addEventListener('update', ...) 接，不是 onmessage
记得调用 eventSource.close() 清理，比如组件卸载时

let es = null;
function connect() {
  es = new EventSource('/sse');
  es.onmessage = e => console.log('data:', e.data);
  es.onerror = () => {
    if (es && es.readyState === 0) {
      setTimeout(connect, 2000);
    }
  };
}
connect();