1. 流式传输到底在解决什么问题第一次接触流式传输这个概念很多人会以为它是什么高深的新技术。其实你每天都在用它只是没意识到而已。打开ChatGPT看它一个字一个字往外蹦回答用手机看直播画面实时传过来甚至你在终端里跑一个命令看日志一行行刷出来背后都是同一套思路在支撑。传统的数据传输方式是什么样客户端发一个请求服务端把全部数据准备好一次性打包返回客户端收到完整响应后再解析、再渲染。这种方式在处理小数据量时没问题但一旦遇到大语言模型生成回答这种场景就完蛋了。模型生成一段500字的回答可能需要十几秒如果等全部生成完再返回用户就要对着空白屏幕干等十几秒体验极差。流式传输的核心思路就一句话不等全部数据准备好生成一点就发一点收到一点就处理一点。这个思路带来的改变是根本性的——首字节到达时间从十几秒缩短到几百毫秒用户感知到的响应速度提升了一个数量级。SSE全称Server-Sent Events是流式传输在Web场景下最常用的一种协议实现。它基于HTTP协议允许服务端主动向客户端推送数据。和WebSocket不同SSE是单向的——只能服务端推给客户端客户端不能通过同一个连接往回发数据。这个限制听起来像是缺点但在很多场景下反而是优势实现更简单不需要额外的协议升级握手走标准HTTP端口兼容性好得多。这篇文章适合谁看如果你正在做AI应用开发需要把大模型的输出实时展示给用户如果你在做实时监控面板需要服务端持续推送状态更新或者你只是想搞清楚ChatGPT那种打字机效果到底怎么实现的——那这篇内容就是写给你的。我会从协议原理讲到代码实现从参数调优讲到踩坑经验尽量把每个环节都说透。2. 流式传输的核心原理与SSE协议拆解2.1 从HTTP的请求-响应模型说起要理解流式传输得先搞清楚普通HTTP请求的工作方式。HTTP本质上是一个请求-响应协议客户端发起连接发送请求头服务端处理完后返回响应头和响应体然后连接关闭或者保持一段时间复用。关键在于客户端需要知道响应体什么时候结束——通常靠Content-Length头来标识响应体的大小或者用Transfer-Encoding: chunked来表示分块传输。普通模式下服务端必须先把所有数据准备好计算出Content-Length然后一次性发送。这就导致了一个问题如果数据是动态生成的生成过程需要时间那服务端要么等全部生成完再算长度要么就得用分块传输。分块传输编码chunked transfer encoding其实是流式传输的底层基础之一。它允许服务端在不知道最终数据总长度的情况下一块一块地发送数据。每块数据前面有一个十六进制的长度标识最后用一个长度为0的块表示结束。这个机制在HTTP/1.1中就已经标准化了但真正让它大放异彩的是大模型时代的到来。2.2 SSE协议的数据格式长什么样SSE的协议格式极其简单简单到你会怀疑它是不是太简陋了。它的基本单位是事件流event stream每个事件由若干行文本组成行与行之间用换行符分隔事件之间用空行分隔。每一行的格式是field: value支持的字段有四个data消息内容这是最核心的字段event事件类型默认是message可以自定义id事件ID用于断线重连时标识最后收到的事件retry重连时间间隔单位毫秒一个典型的SSE响应长这样data: {content: 你} data: {content: 好} data: {content: }注意每个data行后面跟一个空行这个空行是事件分隔符。如果一条消息内容很长需要多行可以写多个data行客户端会把它们用换行符拼接起来。服务端返回的响应头必须包含这几个关键字段Content-Type: text/event-stream Cache-Control: no-cache Connection: keep-alivetext/event-stream是SSE的MIME类型标识浏览器看到这个类型就知道要按事件流来处理。Cache-Control: no-cache是防止中间层缓存响应内容毕竟流式数据每次都不一样。Connection: keep-alive是保持长连接避免每次推送都重新建立TCP连接。2.3 为什么SSE比WebSocket更适合某些场景很多人一提到实时推送就想到WebSocket觉得SSE是低配版。这个认知其实有偏差。两者解决的是不同的问题选型要看具体需求。WebSocket是全双工协议客户端和服务端可以随时互相发消息。它需要一次HTTP升级握手把协议从HTTP切换到WebSocketws://或wss://。握手完成后双方就进入了一个完全对称的通信模式。这个模式适合聊天室、协同编辑、在线游戏这类需要双向实时交互的场景。SSE是单向的只有服务端能推数据给客户端。但它的优势在于完全基于标准HTTP不需要协议升级不需要额外的端口不需要特殊的代理配置。你现有的HTTP基础设施——负载均衡、CDN、反向代理——基本都能直接支持SSE而WebSocket在这些环节经常需要额外配置。还有一个容易被忽略的点SSE自带断线重连机制。浏览器原生的EventSourceAPI在连接断开后会自动尝试重连并且可以通过Last-Event-ID头把最后收到的事件ID带给服务端让服务端知道从哪里继续。WebSocket要实现同样的功能得自己写一套重连逻辑。从实际项目经验来看如果你的场景是服务端持续推送、客户端只需要接收比如AI对话的流式输出、实时日志推送、股票行情更新SSE是更省事的选择。如果确实需要双向通信再考虑WebSocket。3. 手把手实现一个SSE流式接口3.1 服务端实现以Node.js为例先看一个最基础的服务端实现用Node.js的原生http模块const http require(http); const server http.createServer((req, res) { if (req.url /stream) { res.writeHead(200, { Content-Type: text/event-stream, Cache-Control: no-cache, Connection: keep-alive, Access-Control-Allow-Origin: * }); let count 0; const interval setInterval(() { count; res.write(data: ${JSON.stringify({ index: count, time: Date.now() })}\n\n); if (count 10) { clearInterval(interval); res.write(event: done\ndata: {}\n\n); res.end(); } }, 1000); req.on(close, () { clearInterval(interval); console.log(客户端断开连接); }); } }); server.listen(3000, () { console.log(SSE服务运行在3000端口); });这段代码有几个关键点需要注意。res.write()发送的每条消息必须以\n\n结尾这是事件分隔符少了客户端解析不出来。req.on(close)监听客户端断开事件及时清理定时器否则会造成内存泄漏——这是新手最容易踩的坑之一。如果用Express框架代码会更简洁const express require(express); const app express(); app.get(/stream, (req, res) { res.setHeader(Content-Type, text/event-stream); res.setHeader(Cache-Control, no-cache); res.setHeader(Connection, keep-alive); res.flushHeaders(); const timer setInterval(() { res.write(data: ${JSON.stringify({ msg: 心跳 })}\n\n); }, 3000); req.on(close, () { clearInterval(timer); res.end(); }); }); app.listen(3000);注意res.flushHeaders()这一行。Express默认会缓冲响应头不手动flush的话客户端可能等很久才收到第一个字节流式的意义就没了。这个细节在文档里往往不会强调但实际项目中非常关键。3.2 客户端接收EventSource与fetch两种方式浏览器端接收SSE最直接的方式是用原生的EventSourceconst es new EventSource(http://localhost:3000/stream); es.onmessage (event) { const data JSON.parse(event.data); console.log(收到消息:, data); }; es.addEventListener(done, () { console.log(流结束); es.close(); }); es.onerror (err) { console.error(连接出错:, err); // EventSource会自动重连不需要手动处理 };EventSource用起来确实简单但它有几个硬伤不支持自定义请求头只支持GET请求无法携带请求体。这意味着你没法在请求头里放认证Token也没法用POST发送复杂的查询参数。在实际项目中这几乎是不可接受的。所以现在越来越多的项目改用fetch配合ReadableStream来接收SSEasync function fetchStream() { const response await fetch(http://localhost:3000/stream, { method: POST, headers: { Content-Type: application/json, Authorization: Bearer your-token }, body: JSON.stringify({ query: 你好 }) }); const reader response.body.getReader(); const decoder new TextDecoder(); let buffer ; while (true) { const { done, value } await reader.read(); if (done) break; buffer decoder.decode(value, { stream: true }); const lines buffer.split(\n\n); buffer lines.pop(); // 最后一段可能不完整留到下次处理 for (const line of lines) { if (line.startsWith(data: )) { const data JSON.parse(line.slice(6)); console.log(收到:, data); } } } }这段代码里有个容易出错的地方buffer的处理。TCP传输是流式的一次reader.read()拿到的数据不一定正好是一个完整的事件。可能半个事件在这次的chunk里另外半个在下次。所以必须用缓冲区把不完整的数据攒起来等下一个chunk到了再拼接解析。我见过不少项目在这里出bug表现就是偶尔JSON解析失败原因就是没处理好分片边界。3.3 用Python实现服务端和客户端Python生态里FastAPI对SSE的支持比较友好from fastapi import FastAPI from fastapi.responses import StreamingResponse import asyncio import json app FastAPI() async def event_generator(): for i in range(10): data json.dumps({index: i, message: f第{i}条消息}) yield fdata: {data}\n\n await asyncio.sleep(1) yield event: done\ndata: {}\n\n app.get(/stream) async def stream(): return StreamingResponse( event_generator(), media_typetext/event-stream, headers{ Cache-Control: no-cache, Connection: keep-alive, } )Python客户端用requests库也能接收流式响应import requests import json def consume_stream(): with requests.post( http://localhost:8000/stream, json{query: 你好}, streamTrue ) as resp: buffer for chunk in resp.iter_content(chunk_sizeNone, decode_unicodeTrue): if chunk: buffer chunk while \n\n in buffer: event, buffer buffer.split(\n\n, 1) for line in event.split(\n): if line.startswith(data: ): data json.loads(line[6:]) print(收到:, data)streamTrue这个参数是关键不加的话requests会把整个响应体读完才返回流式就失效了。iter_content的chunk_sizeNone表示不限制每次读取的大小来多少读多少这样延迟最低。4. 生产环境中的关键参数与性能调优4.1 心跳机制为什么必须有怎么加SSE连接是长连接可能维持几分钟甚至几小时。中间经过的代理服务器、负载均衡器、防火墙往往有空闲超时设置如果一段时间没有数据传输它们会主动断开连接。用户看到的现象就是用着用着突然没反应了。解决办法是定期发送心跳。心跳本质上就是一条内容为空的SSE消息目的不是传递数据而是保持连接活跃const heartbeat setInterval(() { res.write(: heartbeat\n\n); }, 15000);注意这里用的是:开头的行这是SSE协议里的注释行客户端收到后会忽略不会触发onmessage事件。用注释行做心跳比发空data更干净不会干扰业务逻辑。心跳间隔设多少合适我的经验是15到30秒。太短了浪费带宽和CPU太长了起不到保活作用。具体值要看你的网络链路上各层设备的超时配置一般Nginx默认的keepalive_timeout是65秒所以30秒以内的心跳是安全的。4.2 缓冲区与背压处理流式传输中有一个容易被忽视的问题生产速度大于消费速度。服务端拼命推数据客户端处理不过来数据在缓冲区里越积越多最终导致内存暴涨或者连接被强制断开。Node.js里可以通过res.write()的返回值来判断缓冲区是否满了const canContinue res.write(data: ${chunk}\n\n); if (!canContinue) { // 缓冲区满了暂停生产 await new Promise(resolve res.once(drain, resolve)); }res.write()返回false表示内部缓冲区已满这时候应该暂停写入等drain事件触发后再继续。这个机制叫背压backpressure是流式编程里的核心概念。不处理背压的代码在低负载时看不出问题一旦并发上来就会出各种诡异故障。4.3 Nginx反向代理的关键配置生产环境里SSE服务前面通常有Nginx做反向代理。默认配置下Nginx会缓冲上游响应这会导致流式数据被攒成一大块才发给客户端流式效果完全丧失。必须显式关闭缓冲location /stream { proxy_pass http://backend; proxy_http_version 1.1; proxy_set_header Connection ; # 关闭缓冲这是关键 proxy_buffering off; proxy_cache off; # 延长超时时间 proxy_read_timeout 3600s; proxy_send_timeout 3600s; # 关闭分块传输的缓冲 chunked_transfer_encoding off; }proxy_buffering off是最关键的一行。不关这个Nginx会把后端发来的数据先缓存起来攒够一定大小或者等响应结束才转发给客户端流式就变成了最后一次性返回。我见过好几个项目在这里卡了很久代码逻辑没问题就是Nginx配置没改。proxy_read_timeout也要调大默认60秒。如果SSE连接超过60秒没有数据传输心跳间隔大于60秒的情况下Nginx会主动断开连接。设成3600秒或者更长比较保险。5. 常见问题排查与实战避坑指南5.1 问题速查表现象可能原因排查方向解决方案客户端收不到任何数据响应头Content-Type不对检查响应头是否为text/event-stream修正Content-Type数据一次性全部到达中间层缓冲未关闭检查Nginx/网关的buffering配置关闭proxy_buffering连接几分钟后自动断开空闲超时检查各层超时设置加心跳调大timeout偶尔JSON解析失败分片边界处理不当检查客户端缓冲区逻辑用buffer拼接完整事件内存持续增长客户端断开后未清理资源检查close事件监听清理定时器和监听器浏览器控制台报CORS错误跨域头缺失检查Access-Control-Allow-Origin添加CORS响应头部分浏览器收不到消息响应被压缩检查Content-Encoding禁用gzip压缩5.2 三个最容易踩的坑第一个坑忘了处理客户端断开。服务端在推送数据时如果客户端已经断开了连接res.write()会报错或者静默失败。如果不监听close事件清理定时器、数据库连接、订阅关系这些资源就会一直挂着时间长了就是内存泄漏。我的习惯是每个SSE接口都写一个cleanup函数在close事件里统一调用。第二个坑gzip压缩把流式压没了。有些中间件默认开启gzip压缩而gzip是块压缩算法需要攒够一定数据才能输出压缩块。结果就是流式数据被压缩中间件缓冲了客户端等半天才收到一大坨。解决办法是在SSE接口的响应头里加Content-Encoding: identity或者在该路由上禁用压缩中间件。第三个坑EventSource的自动重连导致重复请求。EventSource在连接断开后会自动重连如果你的服务端逻辑没有做幂等处理重连后可能会重复执行某些操作。比如AI对话场景重连后模型可能从头开始生成用户看到重复内容。解决办法是用Last-Event-ID机制服务端记录每个客户端的进度重连时从断点继续。5.3 调试SSE的实用技巧调试SSE接口浏览器开发者工具是最直接的。在Network面板里找到对应的请求看Response标签页如果配置正确应该能看到数据一行行实时出现。如果数据是突然全部出现的说明中间有缓冲。命令行下用curl也很方便curl -N -H Accept: text/event-stream http://localhost:3000/stream-N参数是禁用curl自己的缓冲不加的话curl也会攒着数据一起输出让你误以为服务端没有流式返回。还有一个技巧是用curl -v看完整的请求响应头确认Content-Type、Transfer-Encoding这些关键头是否正确。很多时候问题就出在某个头不对看一眼就清楚了。6. 从SSE延伸出去的几个实用方向搞懂了SSE之后你会发现很多场景都可以用它来优化。比如后台任务进度推送——用户提交一个耗时任务服务端用SSE实时推送进度百分比比轮询优雅得多也没有轮询的延迟和无效请求。再比如实时日志查看器运维平台里查看某个服务的日志用SSE推送比WebSocket更轻量而且天然支持多标签页同时查看。还有一个方向是多路复用。一个SSE连接可以推送多种类型的事件通过event字段区分。比如同时推送日志、指标、告警三种数据客户端用addEventListener分别监听。这样只需要维持一个连接比开多个连接节省资源。在AI应用开发里SSE几乎成了标配。大模型的流式输出、Agent执行过程的步骤推送、RAG检索到的文档片段实时展示都是用SSE来实现的。掌握SSE的细节对于做AI应用的前端和后端开发来说已经从加分项变成了必备技能。我在实际项目里最大的体会是SSE的协议本身很简单难的是周边设施的配合。代理配置、超时设置、缓冲控制、断线重连、资源清理这些工程细节才是决定流式体验好坏的关键。协议看半小时就懂了但这些坑得一个个踩过来才能记住。希望这篇内容能帮你少走一些弯路。 SSE流式传输实战：从协议原理到生产环境性能调优
1. 流式传输到底在解决什么问题第一次接触流式传输这个概念很多人会以为它是什么高深的新技术。其实你每天都在用它只是没意识到而已。打开ChatGPT看它一个字一个字往外蹦回答用手机看直播画面实时传过来甚至你在终端里跑一个命令看日志一行行刷出来背后都是同一套思路在支撑。传统的数据传输方式是什么样客户端发一个请求服务端把全部数据准备好一次性打包返回客户端收到完整响应后再解析、再渲染。这种方式在处理小数据量时没问题但一旦遇到大语言模型生成回答这种场景就完蛋了。模型生成一段500字的回答可能需要十几秒如果等全部生成完再返回用户就要对着空白屏幕干等十几秒体验极差。流式传输的核心思路就一句话不等全部数据准备好生成一点就发一点收到一点就处理一点。这个思路带来的改变是根本性的——首字节到达时间从十几秒缩短到几百毫秒用户感知到的响应速度提升了一个数量级。SSE全称Server-Sent Events是流式传输在Web场景下最常用的一种协议实现。它基于HTTP协议允许服务端主动向客户端推送数据。和WebSocket不同SSE是单向的——只能服务端推给客户端客户端不能通过同一个连接往回发数据。这个限制听起来像是缺点但在很多场景下反而是优势实现更简单不需要额外的协议升级握手走标准HTTP端口兼容性好得多。这篇文章适合谁看如果你正在做AI应用开发需要把大模型的输出实时展示给用户如果你在做实时监控面板需要服务端持续推送状态更新或者你只是想搞清楚ChatGPT那种打字机效果到底怎么实现的——那这篇内容就是写给你的。我会从协议原理讲到代码实现从参数调优讲到踩坑经验尽量把每个环节都说透。2. 流式传输的核心原理与SSE协议拆解2.1 从HTTP的请求-响应模型说起要理解流式传输得先搞清楚普通HTTP请求的工作方式。HTTP本质上是一个请求-响应协议客户端发起连接发送请求头服务端处理完后返回响应头和响应体然后连接关闭或者保持一段时间复用。关键在于客户端需要知道响应体什么时候结束——通常靠Content-Length头来标识响应体的大小或者用Transfer-Encoding: chunked来表示分块传输。普通模式下服务端必须先把所有数据准备好计算出Content-Length然后一次性发送。这就导致了一个问题如果数据是动态生成的生成过程需要时间那服务端要么等全部生成完再算长度要么就得用分块传输。分块传输编码chunked transfer encoding其实是流式传输的底层基础之一。它允许服务端在不知道最终数据总长度的情况下一块一块地发送数据。每块数据前面有一个十六进制的长度标识最后用一个长度为0的块表示结束。这个机制在HTTP/1.1中就已经标准化了但真正让它大放异彩的是大模型时代的到来。2.2 SSE协议的数据格式长什么样SSE的协议格式极其简单简单到你会怀疑它是不是太简陋了。它的基本单位是事件流event stream每个事件由若干行文本组成行与行之间用换行符分隔事件之间用空行分隔。每一行的格式是field: value支持的字段有四个data消息内容这是最核心的字段event事件类型默认是message可以自定义id事件ID用于断线重连时标识最后收到的事件retry重连时间间隔单位毫秒一个典型的SSE响应长这样data: {content: 你} data: {content: 好} data: {content: }注意每个data行后面跟一个空行这个空行是事件分隔符。如果一条消息内容很长需要多行可以写多个data行客户端会把它们用换行符拼接起来。服务端返回的响应头必须包含这几个关键字段Content-Type: text/event-stream Cache-Control: no-cache Connection: keep-alivetext/event-stream是SSE的MIME类型标识浏览器看到这个类型就知道要按事件流来处理。Cache-Control: no-cache是防止中间层缓存响应内容毕竟流式数据每次都不一样。Connection: keep-alive是保持长连接避免每次推送都重新建立TCP连接。2.3 为什么SSE比WebSocket更适合某些场景很多人一提到实时推送就想到WebSocket觉得SSE是低配版。这个认知其实有偏差。两者解决的是不同的问题选型要看具体需求。WebSocket是全双工协议客户端和服务端可以随时互相发消息。它需要一次HTTP升级握手把协议从HTTP切换到WebSocketws://或wss://。握手完成后双方就进入了一个完全对称的通信模式。这个模式适合聊天室、协同编辑、在线游戏这类需要双向实时交互的场景。SSE是单向的只有服务端能推数据给客户端。但它的优势在于完全基于标准HTTP不需要协议升级不需要额外的端口不需要特殊的代理配置。你现有的HTTP基础设施——负载均衡、CDN、反向代理——基本都能直接支持SSE而WebSocket在这些环节经常需要额外配置。还有一个容易被忽略的点SSE自带断线重连机制。浏览器原生的EventSourceAPI在连接断开后会自动尝试重连并且可以通过Last-Event-ID头把最后收到的事件ID带给服务端让服务端知道从哪里继续。WebSocket要实现同样的功能得自己写一套重连逻辑。从实际项目经验来看如果你的场景是服务端持续推送、客户端只需要接收比如AI对话的流式输出、实时日志推送、股票行情更新SSE是更省事的选择。如果确实需要双向通信再考虑WebSocket。3. 手把手实现一个SSE流式接口3.1 服务端实现以Node.js为例先看一个最基础的服务端实现用Node.js的原生http模块const http require(http); const server http.createServer((req, res) { if (req.url /stream) { res.writeHead(200, { Content-Type: text/event-stream, Cache-Control: no-cache, Connection: keep-alive, Access-Control-Allow-Origin: * }); let count 0; const interval setInterval(() { count; res.write(data: ${JSON.stringify({ index: count, time: Date.now() })}\n\n); if (count 10) { clearInterval(interval); res.write(event: done\ndata: {}\n\n); res.end(); } }, 1000); req.on(close, () { clearInterval(interval); console.log(客户端断开连接); }); } }); server.listen(3000, () { console.log(SSE服务运行在3000端口); });这段代码有几个关键点需要注意。res.write()发送的每条消息必须以\n\n结尾这是事件分隔符少了客户端解析不出来。req.on(close)监听客户端断开事件及时清理定时器否则会造成内存泄漏——这是新手最容易踩的坑之一。如果用Express框架代码会更简洁const express require(express); const app express(); app.get(/stream, (req, res) { res.setHeader(Content-Type, text/event-stream); res.setHeader(Cache-Control, no-cache); res.setHeader(Connection, keep-alive); res.flushHeaders(); const timer setInterval(() { res.write(data: ${JSON.stringify({ msg: 心跳 })}\n\n); }, 3000); req.on(close, () { clearInterval(timer); res.end(); }); }); app.listen(3000);注意res.flushHeaders()这一行。Express默认会缓冲响应头不手动flush的话客户端可能等很久才收到第一个字节流式的意义就没了。这个细节在文档里往往不会强调但实际项目中非常关键。3.2 客户端接收EventSource与fetch两种方式浏览器端接收SSE最直接的方式是用原生的EventSourceconst es new EventSource(http://localhost:3000/stream); es.onmessage (event) { const data JSON.parse(event.data); console.log(收到消息:, data); }; es.addEventListener(done, () { console.log(流结束); es.close(); }); es.onerror (err) { console.error(连接出错:, err); // EventSource会自动重连不需要手动处理 };EventSource用起来确实简单但它有几个硬伤不支持自定义请求头只支持GET请求无法携带请求体。这意味着你没法在请求头里放认证Token也没法用POST发送复杂的查询参数。在实际项目中这几乎是不可接受的。所以现在越来越多的项目改用fetch配合ReadableStream来接收SSEasync function fetchStream() { const response await fetch(http://localhost:3000/stream, { method: POST, headers: { Content-Type: application/json, Authorization: Bearer your-token }, body: JSON.stringify({ query: 你好 }) }); const reader response.body.getReader(); const decoder new TextDecoder(); let buffer ; while (true) { const { done, value } await reader.read(); if (done) break; buffer decoder.decode(value, { stream: true }); const lines buffer.split(\n\n); buffer lines.pop(); // 最后一段可能不完整留到下次处理 for (const line of lines) { if (line.startsWith(data: )) { const data JSON.parse(line.slice(6)); console.log(收到:, data); } } } }这段代码里有个容易出错的地方buffer的处理。TCP传输是流式的一次reader.read()拿到的数据不一定正好是一个完整的事件。可能半个事件在这次的chunk里另外半个在下次。所以必须用缓冲区把不完整的数据攒起来等下一个chunk到了再拼接解析。我见过不少项目在这里出bug表现就是偶尔JSON解析失败原因就是没处理好分片边界。3.3 用Python实现服务端和客户端Python生态里FastAPI对SSE的支持比较友好from fastapi import FastAPI from fastapi.responses import StreamingResponse import asyncio import json app FastAPI() async def event_generator(): for i in range(10): data json.dumps({index: i, message: f第{i}条消息}) yield fdata: {data}\n\n await asyncio.sleep(1) yield event: done\ndata: {}\n\n app.get(/stream) async def stream(): return StreamingResponse( event_generator(), media_typetext/event-stream, headers{ Cache-Control: no-cache, Connection: keep-alive, } )Python客户端用requests库也能接收流式响应import requests import json def consume_stream(): with requests.post( http://localhost:8000/stream, json{query: 你好}, streamTrue ) as resp: buffer for chunk in resp.iter_content(chunk_sizeNone, decode_unicodeTrue): if chunk: buffer chunk while \n\n in buffer: event, buffer buffer.split(\n\n, 1) for line in event.split(\n): if line.startswith(data: ): data json.loads(line[6:]) print(收到:, data)streamTrue这个参数是关键不加的话requests会把整个响应体读完才返回流式就失效了。iter_content的chunk_sizeNone表示不限制每次读取的大小来多少读多少这样延迟最低。4. 生产环境中的关键参数与性能调优4.1 心跳机制为什么必须有怎么加SSE连接是长连接可能维持几分钟甚至几小时。中间经过的代理服务器、负载均衡器、防火墙往往有空闲超时设置如果一段时间没有数据传输它们会主动断开连接。用户看到的现象就是用着用着突然没反应了。解决办法是定期发送心跳。心跳本质上就是一条内容为空的SSE消息目的不是传递数据而是保持连接活跃const heartbeat setInterval(() { res.write(: heartbeat\n\n); }, 15000);注意这里用的是:开头的行这是SSE协议里的注释行客户端收到后会忽略不会触发onmessage事件。用注释行做心跳比发空data更干净不会干扰业务逻辑。心跳间隔设多少合适我的经验是15到30秒。太短了浪费带宽和CPU太长了起不到保活作用。具体值要看你的网络链路上各层设备的超时配置一般Nginx默认的keepalive_timeout是65秒所以30秒以内的心跳是安全的。4.2 缓冲区与背压处理流式传输中有一个容易被忽视的问题生产速度大于消费速度。服务端拼命推数据客户端处理不过来数据在缓冲区里越积越多最终导致内存暴涨或者连接被强制断开。Node.js里可以通过res.write()的返回值来判断缓冲区是否满了const canContinue res.write(data: ${chunk}\n\n); if (!canContinue) { // 缓冲区满了暂停生产 await new Promise(resolve res.once(drain, resolve)); }res.write()返回false表示内部缓冲区已满这时候应该暂停写入等drain事件触发后再继续。这个机制叫背压backpressure是流式编程里的核心概念。不处理背压的代码在低负载时看不出问题一旦并发上来就会出各种诡异故障。4.3 Nginx反向代理的关键配置生产环境里SSE服务前面通常有Nginx做反向代理。默认配置下Nginx会缓冲上游响应这会导致流式数据被攒成一大块才发给客户端流式效果完全丧失。必须显式关闭缓冲location /stream { proxy_pass http://backend; proxy_http_version 1.1; proxy_set_header Connection ; # 关闭缓冲这是关键 proxy_buffering off; proxy_cache off; # 延长超时时间 proxy_read_timeout 3600s; proxy_send_timeout 3600s; # 关闭分块传输的缓冲 chunked_transfer_encoding off; }proxy_buffering off是最关键的一行。不关这个Nginx会把后端发来的数据先缓存起来攒够一定大小或者等响应结束才转发给客户端流式就变成了最后一次性返回。我见过好几个项目在这里卡了很久代码逻辑没问题就是Nginx配置没改。proxy_read_timeout也要调大默认60秒。如果SSE连接超过60秒没有数据传输心跳间隔大于60秒的情况下Nginx会主动断开连接。设成3600秒或者更长比较保险。5. 常见问题排查与实战避坑指南5.1 问题速查表现象可能原因排查方向解决方案客户端收不到任何数据响应头Content-Type不对检查响应头是否为text/event-stream修正Content-Type数据一次性全部到达中间层缓冲未关闭检查Nginx/网关的buffering配置关闭proxy_buffering连接几分钟后自动断开空闲超时检查各层超时设置加心跳调大timeout偶尔JSON解析失败分片边界处理不当检查客户端缓冲区逻辑用buffer拼接完整事件内存持续增长客户端断开后未清理资源检查close事件监听清理定时器和监听器浏览器控制台报CORS错误跨域头缺失检查Access-Control-Allow-Origin添加CORS响应头部分浏览器收不到消息响应被压缩检查Content-Encoding禁用gzip压缩5.2 三个最容易踩的坑第一个坑忘了处理客户端断开。服务端在推送数据时如果客户端已经断开了连接res.write()会报错或者静默失败。如果不监听close事件清理定时器、数据库连接、订阅关系这些资源就会一直挂着时间长了就是内存泄漏。我的习惯是每个SSE接口都写一个cleanup函数在close事件里统一调用。第二个坑gzip压缩把流式压没了。有些中间件默认开启gzip压缩而gzip是块压缩算法需要攒够一定数据才能输出压缩块。结果就是流式数据被压缩中间件缓冲了客户端等半天才收到一大坨。解决办法是在SSE接口的响应头里加Content-Encoding: identity或者在该路由上禁用压缩中间件。第三个坑EventSource的自动重连导致重复请求。EventSource在连接断开后会自动重连如果你的服务端逻辑没有做幂等处理重连后可能会重复执行某些操作。比如AI对话场景重连后模型可能从头开始生成用户看到重复内容。解决办法是用Last-Event-ID机制服务端记录每个客户端的进度重连时从断点继续。5.3 调试SSE的实用技巧调试SSE接口浏览器开发者工具是最直接的。在Network面板里找到对应的请求看Response标签页如果配置正确应该能看到数据一行行实时出现。如果数据是突然全部出现的说明中间有缓冲。命令行下用curl也很方便curl -N -H Accept: text/event-stream http://localhost:3000/stream-N参数是禁用curl自己的缓冲不加的话curl也会攒着数据一起输出让你误以为服务端没有流式返回。还有一个技巧是用curl -v看完整的请求响应头确认Content-Type、Transfer-Encoding这些关键头是否正确。很多时候问题就出在某个头不对看一眼就清楚了。6. 从SSE延伸出去的几个实用方向搞懂了SSE之后你会发现很多场景都可以用它来优化。比如后台任务进度推送——用户提交一个耗时任务服务端用SSE实时推送进度百分比比轮询优雅得多也没有轮询的延迟和无效请求。再比如实时日志查看器运维平台里查看某个服务的日志用SSE推送比WebSocket更轻量而且天然支持多标签页同时查看。还有一个方向是多路复用。一个SSE连接可以推送多种类型的事件通过event字段区分。比如同时推送日志、指标、告警三种数据客户端用addEventListener分别监听。这样只需要维持一个连接比开多个连接节省资源。在AI应用开发里SSE几乎成了标配。大模型的流式输出、Agent执行过程的步骤推送、RAG检索到的文档片段实时展示都是用SSE来实现的。掌握SSE的细节对于做AI应用的前端和后端开发来说已经从加分项变成了必备技能。我在实际项目里最大的体会是SSE的协议本身很简单难的是周边设施的配合。代理配置、超时设置、缓冲控制、断线重连、资源清理这些工程细节才是决定流式体验好坏的关键。协议看半小时就懂了但这些坑得一个个踩过来才能记住。希望这篇内容能帮你少走一些弯路。