Python实现SSE实时推送方法

本文深入剖析了在 Python 中实现可靠 Server-Sent Events（SSE）实时推送的关键实践与避坑指南，明确指出 Flask 因其同步 WSGI 架构无法胜任长连接场景，极易导致连接卡顿、事件丢失和断连残留；强烈推荐使用原生支持异步与连接生命周期管理的 Starlette（轻量可控，适合纯 SSE 服务）或 FastAPI（集成鉴权、校验与 OpenAPI，适合复杂生产环境），并系统讲解了响应头设置、消息格式规范（必须双换行分隔、正确转义换行符、合理使用 id/event/data 字段）、Nginx 缓冲配置、以及最关键的客户端断连处理——通过捕获 ClientDisconnect 异常并在 finally 块中彻底清理后台任务与资源，避免内存泄漏与句柄耗尽，真正实现“推得稳、断得净、收得干净”。

用 Starlette 或 FastAPI 实现 SSE，别碰 Flask

Python 原生不支持长连接流式响应的优雅生命周期管理，Flask 的 yield + Response 表面能跑，但实际会卡住连接、丢事件、无法处理客户端断连。生产环境必须用异步框架。

原因很简单：Flask 是同步 WSGI 框架，每个请求占一个线程/进程，SSE 要求连接长期挂起，而 WSGI 服务器（如 gunicorn）默认超时 30 秒，且无法感知客户端是否已断开；Starlette 和 FastAPI 基于 ASGI，原生支持协程级连接维持和断连检测。

• Starlette 更轻量，StreamingResponse 接口直接、可控性强，适合纯 SSE 服务
• FastAPI 自带 OpenAPI 和依赖注入，适合需要鉴权、参数校验、与其他 API 共存的场景
• 避免用 aiohttp.web：它没有内置的 SSE 响应包装，要手动写 Content-Type、Cache-Control、text/event-stream 分隔符，容易漏掉 retry: 或换行格式，导致浏览器重连失败

StreamingResponse 返回前必须设对响应头

浏览器只认 Content-Type: text/event-stream，且要求响应头在流开始前全部发出。少一个头，或者晚发，都会让 EventSource 卡在 connecting 状态，不报错也不触发 message。

常见错误现象：EventSource 打开后一直 pending，Network 面板里状态是 (pending)，控制台无报错。

• 必须显式设置 headers={"Content-Type": "text/event-stream", "Cache-Control": "no-cache", "Connection": "keep-alive"}
• Cache-Control 不设或设错（比如写成 no-store），某些浏览器（尤其是 Safari）会拒绝解析流
• 不要依赖框架自动加 Transfer-Encoding: chunked —— ASGI 服务器（如 uvicorn）通常会加，但如果你套了 Nginx，得确认它没缓冲响应；加 proxy_buffering off; 到 location 块里

发送事件时每条消息结尾必须是 \n\n，不是 \n

SSE 协议规定：每条事件以空行（\n\n）分隔。少一个 \n，后续所有事件都会被浏览器拼成一条，data: 字段解析错乱，id: 和 event: 失效。

典型错误写法：yield "data: hello\n" → 浏览器收不到 message 事件；正确是 yield "data: hello\n\n"。

• 完整单条事件示例："id: 123\nevent: update\ndata: {\"status\":\"ok\"}\n\n"
• id: 字段用于断线重连后的事件去重，生产环境建议生成单调递增或时间戳 ID
• 如果数据含换行符（比如 JSON 字符串），必须对 data: 行做转义：把 \n 替换成 \\n，否则会提前触发分隔

客户端断连后，服务端协程得真正退出，别留僵尸任务

ASGI 应用里，StreamingResponse 的生成器函数一旦被客户端关闭，uvicorn 会抛 ClientDisconnect 异常。但如果你在生成器里开了后台任务（比如 asyncio.create_task()）、用了未取消的定时器、或忘了 try/except 捕获，协程就卡在 await 上，连接释放了，任务还在跑。

后果：内存缓慢增长，fd 耗尽，uvicorn 日志出现 Task was destroyed but it is pending!。

• 务必在生成器最外层包 try...except starlette.exceptions.ClientDisconnect:，并在 finally 里清理资源
• 如果用了 asyncio.sleep() 循环推送，改用 asyncio.wait_for(..., timeout=...) 并捕获 asyncio.TimeoutError 和 ClientDisconnect
• 避免在 SSE 路由里调用阻塞 IO（如 time.sleep、requests.get），它们会阻塞整个 event loop

真正难的不是推数据，是推完还能干净收手。很多线上问题都出在“以为连接断了，其实任务还挂着”。

好了，本文到此结束，带大家了解了《Python实现SSE实时推送方法》，希望本文对你有所帮助！关注golang学习网公众号，给大家分享更多文章知识！