如何用Python实现SSE实时推送-Python教程-PHP中文网

sse是基于http的单向服务器推送技术，websocket是需握手升级的双向全双工协议；前者轻量、自动重连、兼容http设施，后者低延迟、支持二进制、适合高频交互。

如何用python实现sse实时推送

什么是SSE，它和WebSocket有什么根本区别 SSE（Server-Sent Events）是单向、服务端主动推送的 HTTP 协议机制，浏览器原生支持，不需要额外库。它本质是长连接的响应流：服务器保持 `Content-Type: text/event-stream` 响应打开，持续写入以 `data:` 开头的事件行。而 WebSocket 是双向全双工协议，需握手升级、手动管理连接状态，适合高频交互（如聊天），但复杂度高、CDN/代理兼容性差。如果你只需要「服务端发、前端收」的实时通知（如日志流、状态更新、进度广播），SSE 更轻量、更易部署、天然支持自动重连。

Flask + SSE 的最小可行实现关键不是“怎么启动一个流”，而是“如何让响应不结束、不缓冲、不被中间件截断”。常见错误包括：Response 被 Flask 自动关闭、WSGI 服务器（如 Gunicorn）默认启用响应缓冲、Nginx 默认关闭长连接支持。

使用 yield 生成器函数返回 Response，并显式设置 content_type="text/event-stream"
关闭 Flask 的响应自动编码：在 Response 构造时传入 direct_passthrough=True
每次推送后调用 sys.stdout.flush() 或 response.stream.flush()（取决于封装方式），防止 WSGI 缓冲
在生成器中插入空注释行 ":\n"，维持连接心跳，避免代理超时断开

from flask import Flask, Response
import time
import sys
<p>app = Flask(<strong>name</strong>)</p><div class="aritcle_card flexRow artxards">
											<div class="artcardd flexRow">
												<a class="aritcle_card_img" rel="nofollow" href="/xiazai/skill2969" title="python-docx"><img
														src="https://img.php.cn/upload/skill/000/000/081/178943532695530.jpg" alt="python-docx" onerror="this.onerror='';this.src='/static/lhimages/moren/morentu.png'" ></a>
												<div class="aritcle_card_info flexColumn">
													<a rel="nofollow" href="/xiazai/skill2969" title="python-docx" class="overflowclass">python-docx</a>
													<p class="overflowclass">python-docx Skill功能概述python-docx Skill是一项面向实际任务的技能，主要用于本Skill提供使用python-docx生成专业Word文档的标准方法和最佳实践；生成安全服务方案文档；核心要点生成技术架构设计文档；生成任何需要专业排版的Word文档；核心库 : python-docx；使用与执行辅助库 : docx.shared , docx.enum , docx.oxml.ns；标准代码模板；1. 文档初始化；2. 字体设置（必须！它将相关步骤、工具调用和结果整理方式集</p>
												</div>
												<a rel="nofollow" href="/xiazai/skill2969" title="python-docx" class="aritcle_card_btn flexRow flexcenter"><b></b><span>下载</span>
												</a>
											</div>
										</div><p>def event_stream():
for i in range(5):
yield f"data: {{\"count\": {i}}}\n\n"
time.sleep(1)
yield ": keep-alive\n"</p><p>@app.route('/stream')
def stream():
return Response(event_stream(), content_type='text/event-stream', direct_passthrough=True)</p>

前端如何正确接收并处理SSE连接浏览器 EventSource API 看似简单，但几个细节极易出错：

EventSource 只接受 GET 请求，且无法携带自定义 headers（如认证 token），必须用 query 参数或 cookie 传参
连接失败时不会抛异常，而是触发 onerror 回调，但该回调可能被反复触发（如网络抖动），需自行节流或判断状态
默认重连间隔为 3 秒，可通过服务端 retry: 字段覆盖，例如 retry: 5000\n
推送的 data: 内容会被自动拼接（遇到空行才触发 onmessage），所以 JSON 字符串不能换行，否则解析失败

const es = new EventSource("/stream?token=abc");
es.onmessage = (e) => {
  const data = JSON.parse(e.data);
  console.log(data.count);
};
es.onerror = (err) => {
  console.warn("SSE connection error", err);
};

Nginx 和 Gunicorn 的必要配置项本地开发跑通不代表线上可用。Gunicorn 默认使用 sync worker，会阻塞整个进程；Nginx 默认关闭长连接、限制响应头大小、启用缓冲——三者叠加直接导致 SSE 失效。

Gunicorn 启动加参数：--worker-class gevent --workers 2 --timeout 120（gevent 支持异步流，timeout 防止连接被强杀）

Nginx 配置中，对应 location 块必须包含：

proxy_cache off;
proxy_buffering off;
proxy_http_version 1.1;
proxy_set_header Upgrade $http_upgrade;
proxy_set_header Connection 'upgrade';
proxy_read_timeout 180;