Python后端发SSE需用流式响应(Flask用Response+生成器,FastAPI用StreamingResponse),设Content-Type: text/event-stream,每条消息以data:开头、双换行结尾,禁用缓存与缓冲;前端用EventSource监听,注意同源与自动重连机制。

Python 后端怎么发 SSE 流(Flask/FastAPI)
直接用 yield 生成响应体,配合正确的 Content-Type 和缓存控制头,就能让浏览器持续接收事件。关键不是“写个接口”,而是让 HTTP 连接不关闭、数据按 SSE 格式分块推送。
常见错误是返回普通 JSON 或用了 jsonify —— 那只是单次响应,浏览器收完就断了,根本不是流。
- 必须设
Content-Type: text/event-stream,否则浏览器不识别为 SSE - 响应体每条消息以
data:开头,结尾加双换行(\n\n),可选id:、event:字段 - 避免框架自动加
Content-Length(比如 Flask 默认会算),得手动禁用或用流式响应对象 - FastAPI 要用
StreamingResponse+yield;Flask 用Response+ 生成器,且需关掉make_response的自动包装
示例(Flask):
from flask import Response, stream_with_context
<p>@app.route('/events')
def sse_stream():
def event_generator():
for i in range(5):
yield f"data: {{\"count\": {i}}}\n\n"
time.sleep(1) # 模拟延迟
return Response(
stream_with_context(event_generator()),
content_type='text/event-stream',
headers={'Cache-Control': 'no-cache', 'X-Accel-Buffering': 'no'}
)前端怎么接 Python 推的 SSE(EventSource 基本用法)
EventSource 是浏览器原生 API,它只认 text/event-stream,不支持 POST、带 body 的请求,也不走 fetch 那套逻辑 —— 这点容易踩坑。
立即学习“Python免费学习笔记(深入)”;
- 地址必须同源或服务端配好
Access-Control-Allow-Origin,否则连连接都建不起来 -
onmessage只收message类型事件(没event:字段的),有自定义类型要用addEventListener('your-event-name', ...) - 连接断开后浏览器默认 3 秒重连,但重连时会带上上次收到的
Last-Event-ID(前提是后端写了id:) - 别在
onerror里直接new EventSource(...)手动重连 —— 容易雪崩,优先靠浏览器自动机制
简单接法:
const es = new EventSource('/events');
es.onmessage = (e) => {
console.log(JSON.parse(e.data));
};
es.addEventListener('error', (e) => {
console.warn('SSE connection error', e);
});为什么本地调试时 Chrome 显示 pending 就不动了
不是代码卡住,是浏览器在等完整响应块 —— SSE 要求每条消息以两个换行结束,少一个 \n,或者服务端没 flush 缓冲区,Chrome 就一直挂起。
- Python 默认 stdout/stderr 行缓冲,但 Web 框架常把响应体全缓存再发;Flask 需加
stream_with_context,FastAPI 要确保StreamingResponse的media_type正确 - 某些反向代理(如 Nginx)默认缓冲响应,要加配置:
proxy_buffering off;、proxy_cache off;、proxy_http_version 1.1; - 开发时用
curl -N http://localhost:5000/events看原始输出最准,能立刻发现换行或头缺失问题
要不要用 async/await 写 SSE(FastAPI vs Flask)
要看数据来源。如果只是定时 yield 静态数据,同步生成器完全够用;但如果要等数据库查询、HTTP 调用或长任务结果,就得异步——但注意:不是加个 async def 就行。
- FastAPI 的
StreamingResponse不直接支持async generator(v0.104+ 才开始实验性支持),稳妥做法是用asyncio.Queue或线程池桥接 - Flask 原生不支持 async route,硬上得用
flask-socketio或换框架,得不偿失 - 真正需要异步的场景其实是“多客户端共享一个数据源”(比如监听 Redis Pub/Sub),这时后端该用单独 consumer 线程/进程推数据到内存队列,SSE 接口只负责从队列
yield
多数内部管理后台的实时日志、状态轮询,同步流更稳、更容易 debug。
真实项目里最难的往往不是怎么发,而是怎么安全地让多个请求复用同一个数据源,又不互相阻塞 —— 这部分没有标准答案,得看你的数据更新节奏和并发量。


















