50 行 SSE + 打字机效果：从零跑通服务端推流 Demo

50 行 SSE 打字机效果从零跑通服务端推流 Demo复制本文代码即可运行。效果浏览器打开页面后服务端每 5 秒推送一轮回复文字逐段流式出现类似 AI 对话打字机。你将得到什么一条SSE 长连接GET /sse-stream服务端把整句拆成delta片段推送前端收到一段、追加一段实现打字机心跳保活 客户端断开清理运行效果示意[11:21:15] 服务端定时推送第1条消息这是打字机效果演示。 [11:21:20] 服务端定时推送第2条消息这是打字机效果演示。 ...一、创建项目mkdirsse-democdsse-demonpminit-ynpminstallexpress目录结构sse-demo/ ├── package.json ├── server.js └── index.html二、package.json{name:sse-demo,version:1.0.0,scripts:{dev:node server.js},dependencies:{express:^5.2.1}}三、server.js完整可复制constexpressrequire(express);constappexpress();constport3000;// 静态托管前端页面index.htmlapp.use(express.static(./));// 模拟一句完整回复拆成逐段 delta 推送functionpushTypewriterMessage(res,num){constfullText服务端定时推送第${num}条消息这是打字机效果演示。;letindex0;consttypeTimersetInterval((){if(indexfullText.length){clearInterval(typeTimer);res.write(data:${JSON.stringify({type:done,id:num})}\n\n);return;}constchunkfullText.slice(index,index2);indexchunk.length;constdataJSON.stringify({type:delta,id:num,delta:chunk,});res.write(data:${data}\n\n);},80);returntypeTimer;}// SSE 接口app.get(/sse-stream,(req,res){// SSE 必须的响应头res.setHeader(Content-Type,text/event-stream);res.setHeader(Cache-Control,no-cache);res.setHeader(Connection,keep-alive);res.setHeader(Access-Control-Allow-Origin,*);// 心跳包防止网关/nginx 超时断开constheartBeatTimersetInterval((){res.write(data: ping\n\n);},10000);// 业务定时推送每 5 秒模拟一轮新回复letnum0;consttypeTimers[];constpushTimersetInterval((){num;consttimerpushTypewriterMessage(res,num);typeTimers.push(timer);},5000);// 客户端关闭连接清除定时器req.on(close,(){console.log(客户端关闭连接清除定时器);clearInterval(heartBeatTimer);clearInterval(pushTimer);typeTimers.forEach(clearInterval);res.end();});});app.listen(port,(){console.log(SSE服务运行在 http://localhost:${port});});四、index.html完整可复制!DOCTYPEhtmlhtmllangzh-CNheadmetacharsetUTF-8titleSSE 实时推送 Demo/titlestylebody{padding:20px;font-size:16px;}#msgList{margin-top:20px;border:1px solid #eee;padding:10px;min-height:300px;max-height:500px;overflow-y:auto;}.item{padding:6px 0;border-bottom:1px solid #f5f5f5;line-height:1.6;}.item .time{color:#666;margin-right:4px;}.ping{color:#999;font-size:12px;}/style/headbodyh3SSE 服务端定时消息实时展示打字机效果/h3dividstatus连接状态未连接/divdividmsgList/divscriptconststatusEldocument.getElementById(status);constlistEldocument.getElementById(msgList);letcurrentLinenull;constssenewEventSource(http://localhost:3000/sse-stream);sse.onopen(){statusEl.textContent连接状态已建立;statusEl.style.colorgreen;};sse.onmessage(e){consttexte.data;if(textping){appendPing();return;}constdataJSON.parse(text);if(data.typedelta){if(!currentLine||currentLine.dataset.id!String(data.id)){currentLinedocument.createElement(div);currentLine.classNameitem;currentLine.dataset.iddata.id;consttimeSpandocument.createElement(span);timeSpan.classNametime;timeSpan.textContent[${newDate().toLocaleTimeString()}];currentLine.appendChild(timeSpan);consttextSpandocument.createElement(span);textSpan.classNametext;currentLine.appendChild(textSpan);listEl.appendChild(currentLine);}currentLine.querySelector(.text).textContentdata.delta;listEl.scrollToplistEl.scrollHeight;return;}if(data.typedone){currentLinenull;}};sse.onerror(){statusEl.textContent连接状态断开/异常正在重连...;statusEl.style.colorred;};functionappendPing(){constdivdocument.createElement(div);div.classNameitem ping;div.innerText[心跳包]${newDate().toLocaleTimeString()};listEl.appendChild(div);listEl.scrollToplistEl.scrollHeight;}// 手动关闭连接// sse.close();/script/body/html五、启动npmrun dev浏览器访问http://localhost:3000若端口被占用先停掉旧进程再启动lsof-ti:3000|xargskill-9npmrun dev六、核心原理1. SSE 消息格式每条消息必须是data: {type:delta,id:1,delta:服务}末尾两个换行\n\n表示一条 event 结束。2. 自定义协议delta donetype含义示例delta增量文本片段{ type:delta, id:1, delta:服务 }done本轮回复结束{ type:done, id:1 }ping心跳纯字符串ping这和 AI 对话很像服务端不断推delta客户端text delta而不是一次推整句。3. 时序浏览器 服务端 │ │ │── GET /sse-stream ─────►│ 建立 SSE 长连接 │◄── delta: 服务 ────────│ 80ms │◄── delta: 端定 ────────│ 80ms │◄── delta: 时推 ────────│ ... │◄── done ─────────────────│ 本轮结束 │ │ 5 秒后下一轮 │◄── delta: 服务 ────────│ id2 新一轮 │◄── ping ─────────────────│ 10s 心跳4. 前端打字机逻辑// 新 id → 新建一行// 同 id → 往 .text 追加 deltacurrentLine.querySelector(.text).textContentdata.delta;// done → 释放 currentLine下一轮 delta 会开新行if(data.typedone)currentLinenull;5. 心跳为什么需要Nginx / 负载均衡对空闲长连接常有 60s 超时。每 10s 推ping避免连接被中间层静默掐断。生产环境更推荐 SSE 注释行: ping\n\n浏览器会自动忽略不进onmessage。6. 断开检测req.on(close,(){clearInterval(heartBeatTimer);clearInterval(pushTimer);typeTimers.forEach(clearInterval);res.end();});关 Tab、刷新、sse.close()都会触发避免服务端定时器泄漏。七、可调参数在server.js的pushTypewriterMessage里参数当前值效果index 2每次 2 字chunk 越大跳字感越强8080ms越小打字越快50005s两轮回复间隔1000010s心跳间隔八、和 AI 对话的对应关系Demo生产如灵犀 SSEdelta字段flowResult.outContent增量doneSSE 连接关闭 finishReason: stopEventSourceGETfetchEventSourcePOST 鉴权textContent deltamarkdownRender.addChunk(delta)这个 Demo 用原生EventSource GET适合理解原理真实 AI 接口一般是POST SSE需要microsoft/fetch-event-source等库。九、常见问题Q看不到打字效果整句一次出来多半是旧版 server 还在跑重启node server.js或在 Network 里确认收到的是{type:delta,...}而不是整句 JSON。Qonerror一直红字EventSource断线会自动重连onerror可能短暂出现一般可恢复。Q想改成 POST SSE原生EventSource不支持 POST需换fetch 流式读 body或microsoft/fetch-event-source。十、小结这个 Demo 覆盖了 SSE 流式 UI 的关键链路服务端长连接 res.write逐段推delta客户端EventSource.onmessage 增量追加工程化心跳保活、req.on(close)清理三个文件、npm run dev、打开浏览器——就能本地复现 AI 式打字机效果。