Server-Sent Events (SSE) 协议SSE 能覆盖 80% 以上的实时场景而且实现简单、资源消耗低是一个被低估的技术。一、SSE 是什么SSEServer-Sent Events是一种基于 HTTP 的轻量级实时通信技术允许服务器主动向客户端推送数据。简单说就是客户端发起一个请求服务器保持连接打开后续有数据就随时推过来。和 WebSocket 的区别SSE单向服务器→客户端协议简单自带重连WebSocket双向协议复杂需要自己处理重连大多数场景其实不需要双向通信SSE 够用了。二、SSE 的使用场景场景说明消息通知新消息提醒、系统通知进度更新文件上传进度、任务处理进度实时日志服务器日志流、构建输出数据推送股票行情、实时天气LLM 流式输出AI 对话逐字显示三、SSE vs 轮询 vs WebSocket对比项SSE轮询WebSocket实时性高低取决于间隔高协议HTTPHTTPWS/WSS方向单向服务端→客户端双向双向自动重连✅ 内置❌ 需手动❌ 需手动消息格式文本UTF-8任意文本/二进制连接数开销小很大频繁建连中等防火墙穿透好HTTP好HTTP部分环境可能被拦浏览器兼容现代浏览器全部现代浏览器实现复杂度极低低较高四、前端代码基础用法consteventSourcenewEventSource(/stream);// 接收普通消息eventSource.onmessagefunction(event){console.log(收到消息:,event.data);};// 接收自定义事件eventSource.addEventListener(progress,function(event){console.log(进度:,event.data);});// 连接错误eventSource.onerrorfunction(event){if(eventSource.readyStateEventSource.CLOSED){console.log(连接关闭);}else{console.error(连接异常);}};关闭连接// 不再需要时关闭eventSource.close();带会话ID的连接functionconnectSSE(sessionId){constesnewEventSource(/stream/${sessionId});es.onmessagefunction(event){// 处理消息document.getElementById(output).innerHTMLevent.databr;};returnes;}任务提交 SSE 监听asyncfunctionsubmitTask(sessionId){// 1. 提交任务awaitfetch(/submit/${sessionId},{method:POST});// 2. 监听结果constesnewEventSource(/stream/${sessionId});es.onmessagefunction(event){constdataJSON.parse(event.data);console.log(data);};}封装成 PromisefunctionwaitForSSEComplete(sessionId,timeout30000){returnnewPromise((resolve,reject){constesnewEventSource(/stream/${sessionId});letresult[];es.onmessagefunction(event){result.push(event.data);};es.addEventListener(complete,function(event){es.close();resolve({messages:result,final:event.data});});es.addEventListener(error,function(event){es.close();reject(newError(连接异常));});setTimeout((){es.close();reject(newError(请求超时));},timeout);});}五、后端代码Python FastAPI基本 SSE 端点fromfastapiimportFastAPIfromfastapi.responsesimportStreamingResponseimportasyncio appFastAPI()asyncdefevent_generator():foriinrange(10):yieldfdata: 进度{i}/10\n\nawaitasyncio.sleep(1)yieldfdata: 完成\n\napp.get(/stream)asyncdefstream():returnStreamingResponse(event_generator(),media_typetext/event-stream)带会话ID的流importjsonfromfastapiimportFastAPIfromfastapi.responsesimportStreamingResponse appFastAPI()tasks{}# 存储任务状态app.get(/stream/{session_id})asyncdefstream_events(session_id:str):defgenerate():# 发送开始消息yieldfdata:{json.dumps({status:started,session_id:session_id})}\n\n# 模拟任务进度foriinrange(1,6):yieldfevent: progress\nyieldfdata:{json.dumps({step:i,total:5})}\n\ntime.sleep(1)# 发送完成消息yieldfevent: complete\nyieldfdata:{json.dumps({result:成功})}\n\nreturnStreamingResponse(generate(),media_typetext/event-stream,headers{Cache-Control:no-cache,Connection:keep-alive,X-Accel-Buffering:no# 禁用nginx缓冲})进度更新的实用写法importasyncioimportjsonfromfastapiimportFastAPI,BackgroundTasksfromfastapi.responsesimportStreamingResponse appFastAPI()# 模拟长时间任务asyncdeflong_task(task_id:str,progress_queue:asyncio.Queue):foriinrange(1,11):awaitasyncio.sleep(1)awaitprogress_queue.put(f进度:{i*10}%)awaitprogress_queue.put(完成)app.post(/submit/{task_id})asyncdefsubmit_task(task_id:str,background_tasks:BackgroundTasks):queueasyncio.Queue()background_tasks.add_task(long_task,task_id,queue)return{status:accepted,task_id:task_id}app.get(/stream/{task_id})asyncdefstream_task(task_id:str):queueasyncio.Queue()asyncdefgenerate():# 检查是否有对应任务的队列简化whileTrue:try:msgawaitasyncio.wait_for(queue.get(),timeout30)yieldfdata:{msg}\n\nifmsg完成:breakexceptasyncio.TimeoutError:yielddata: 心跳\n\nreturnStreamingResponse(generate(),media_typetext/event-stream)六、消息格式说明SSE 协议的消息格式很简单# 普通消息 data: 这是消息内容\n\n # 多行数据 data: 第一行\n data: 第二行\n\n # 带事件类型 event: progress\ndata: 50%\n\n # 注释客户端忽略 : 这是一条注释\n\n # JSON 格式 data: {name: 张三, age: 25}\n\n七、SSE 消息格式速查表写法含义客户端触发的事件data: 内容\n\n普通消息onmessageevent: 名称\ndata: 内容\n\n自定义事件addEventListener(名称, ...): 注释\n\n注释忽略retry: 3000\n\n重连间隔自动生效八、常见问题处理1. 断线重连SSE 自带了重连机制。服务器可以通过retry指令告诉客户端等待多久后重试yieldretry: 3000\n\n# 3秒后重连2. 心跳保持长时间没有消息某些代理/防火墙可能会断开连接。通过定时发送注释或数据来保持连接活跃asyncdefgenerate():whileTrue:yield: heartbeat\n\n# 注释客户端忽略awaitasyncio.sleep(15)3. nginx 缓存问题nginx 默认会缓冲响应需要关闭proxy_buffering off;或者在 FastAPI 中添加响应头headers{X-Accel-Buffering:no}4. 连接数限制SSE 每个客户端占用一个长连接。如果并发高需要注意操作系统文件描述符限制ulimit -n反向代理的连接数配置应用服务器的最大并发连接数九、完整示例AI 对话流式输出!-- frontend --!DOCTYPEhtmlhtmlbodyinputidinputplaceholder输入问题buttononclickask()发送/buttondividoutput/divscriptleteventSourcenull;functionask(){constquestiondocument.getElementById(input).value;if(!question)return;// 关闭旧连接if(eventSource){eventSource.close();}// 提交问题fetch(/chat,{method:POST,headers:{Content-Type:application/json},body:JSON.stringify({question:question})});// 监听流式输出constoutputdocument.getElementById(output);output.innerHTML;eventSourcenewEventSource(/stream/chat);eventSource.onmessagefunction(event){constdataJSON.parse(event.data);if(data.done){eventSource.close();return;}output.innerHTMLdata.text;};}/script/body/html# backendfromfastapiimportFastAPIfromfastapi.responsesimportStreamingResponseimportjsonimportasyncio appFastAPI()app.post(/chat)asyncdefchat(request:dict):questionrequest.get(question,)# 将问题放入处理队列...return{status:ok}app.get(/stream/chat)asyncdefchat_stream():asyncdefgenerate():# 模拟 AI 逐字输出text你好MongoDB 是一个文档型数据库...forcharintext:yieldfdata:{json.dumps({text:char,done:False})}\n\nawaitasyncio.sleep(0.05)yieldfdata:{json.dumps({text:,done:True})}\n\nreturnStreamingResponse(generate(),media_typetext/event-stream)十、使用建议需要实时推送但不需要双向通信 → 优先考虑 SSE不需要兼容 IE 浏览器 → SSE 完全可用服务器资源有限 → SSE 比 WebSocket 更省需要二进制数据 → 用 WebSocket需要客户端也向服务器发消息 → 结合普通 HTTP 请求 SSE