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

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

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();

Flask SSE 长连接卡住、CPU 占高、或者并发撑不住怎么办

Flask 默认开发服务器（Werkzeug）是单线程同步模型，一个 SSE 连接就会占住一个 worker，100 个用户连上来，它就卡死。这不是代码写错了，是部署方式错了。

生产环境必须换异步服务器，否则所有优化都白搭。另外，别在生成器里做耗时操作（比如查数据库、调外部 API），会阻塞整个流。

• 开发调试可用 flask run --reload --no-debugger，但仅限 1–2 人测试
• 生产必须用 gunicorn + gevent 或 uvicorn（搭配 hypercorn 更稳），并显式启用 async workers
• 若需广播（如发消息给所有客户端），别用全局 list 存 connection —— Flask 没共享内存，得用 Redis + Pub/Sub 或消息队列
• 每个 yield 后建议 time.sleep(0.01) 让出控制权，尤其用 gevent 时

为什么 Chrome 控制台看不到 SSE 数据，但 Network 里 status 是 200

因为 Chrome DevTools 的 Network 面板对 text/event-stream 支持有限：它显示连接建立成功（200），但不会实时渲染每条 data: 消息。你得靠 console.log 或 onmessage 打印才能确认是否收到。

另一个高频问题是跨域：如果前端域名和 Flask 后端不一致，EventSource 会静默失败（不触发 onerror），控制台只报 CORS 错误。SSE 不支持带凭证的跨域，除非后端明确允许。

• 务必在 Flask 响应头加 Access-Control-Allow-Origin: *（或具体域名），否则前端连都连不上
• 别用 localhost:3000 连 127.0.0.1:5000 —— 浏览器认为这是两个源
• Chrome 的 Application → Service Workers 面板里也看不到 SSE，它和 SW 无关
• 真要调试流内容，用 curl -N http://localhost:5000/sse 最直接，-N 关闭缓存

Flask SSE 的关键不在语法多难，而在于理解它是长生命周期 HTTP 连接，一环松动（服务模型、响应格式、前端重连、跨域）就会断。很多人卡在“明明代码跑通了却收不到数据”，其实问题早就不在 Python 里了。

今天关于《Flask实现SSE实时推送方法详解》的内容就介绍到这里了，是不是学起来一目了然！想要了解更多关于的内容请关注golang学习网公众号！