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

复制本文代码即可运行。效果：浏览器打开页面后，服务端每 5 秒推送一轮回复，文字逐段流式出现，类似 AI 对话打字机。

你将得到什么

  • 一条 SSE 长连接（GET /sse-stream）
  • 服务端把整句拆成 delta 片段 推送
  • 前端收到一段、追加一段，实现打字机
  • 心跳保活 + 客户端断开清理

运行效果示意：

[11:21:15] 服务端定时推送第1条消息，这是打字机效果演示。
[11:21:20] 服务端定时推送第2条消息，这是打字机效果演示。
...

一、创建项目

mkdir sse-demo && cd sse-demo
npm init -y
npm install express

目录结构：

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（完整可复制）

const express = require('express');
const app = express();
const port = 3000;

// 静态托管前端页面（index.html）
app.use(express.static('./'));

// 模拟一句完整回复，拆成逐段 delta 推送
function pushTypewriterMessage(res, num) {
  const fullText = `服务端定时推送第${num}条消息，这是打字机效果演示。`;
  let index = 0;

  const typeTimer = setInterval(() => {
    if (index >= fullText.length) {
      clearInterval(typeTimer);
      res.write(`data: ${JSON.stringify({ type: 'done', id: num })}\n\n`);
      return;
    }

    const chunk = fullText.slice(index, index + 2);
    index += chunk.length;

    const data = JSON.stringify({
      type: 'delta',
      id: num,
      delta: chunk,
    });
    res.write(`data: ${data}\n\n`);
  }, 80);

  return typeTimer;
}

// 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 超时断开
  const heartBeatTimer = setInterval(() => {
    res.write('data: ping\n\n');
  }, 10000);

  // 业务定时推送：每 5 秒模拟一轮新回复
  let num = 0;
  const typeTimers = [];
  const pushTimer = setInterval(() => {
    num++;
    const timer = pushTypewriterMessage(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（完整可复制）

<!DOCTYPE html>
<html lang="zh-CN">

<head>
    <meta charset="UTF-8">
    <title>SSE 实时推送 Demo</title>
    <style>
        body {
            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>
</head>

<body>
    <h3>SSE 服务端定时消息实时展示（打字机效果）</h3>
    <div id="status">连接状态：未连接</div>
    <div id="msgList"></div>

    <script>
        const statusEl = document.getElementById('status');
        const listEl = document.getElementById('msgList');

        let currentLine = null;

        const sse = new EventSource('http://localhost:3000/sse-stream');

        sse.onopen = () => {
            statusEl.textContent = '连接状态：已建立';
            statusEl.style.color = 'green';
        };

        sse.onmessage = (e) => {
            const text = e.data;
            if (text === 'ping') {
                appendPing();
                return;
            }

            const data = JSON.parse(text);

            if (data.type === 'delta') {
                if (!currentLine || currentLine.dataset.id !== String(data.id)) {
                    currentLine = document.createElement('div');
                    currentLine.className = 'item';
                    currentLine.dataset.id = data.id;

                    const timeSpan = document.createElement('span');
                    timeSpan.className = 'time';
                    timeSpan.textContent = `[${new Date().toLocaleTimeString()}]`;
                    currentLine.appendChild(timeSpan);

                    const textSpan = document.createElement('span');
                    textSpan.className = 'text';
                    currentLine.appendChild(textSpan);

                    listEl.appendChild(currentLine);
                }
                currentLine.querySelector('.text').textContent += data.delta;
                listEl.scrollTop = listEl.scrollHeight;
                return;
            }

            if (data.type === 'done') {
                currentLine = null;
            }
        };

        sse.onerror = () => {
            statusEl.textContent = '连接状态：断开/异常，正在重连...';
            statusEl.style.color = 'red';
        };

        function appendPing() {
            const div = document.createElement('div');
            div.className = 'item ping';
            div.innerText = `[心跳包] ${new Date().toLocaleTimeString()}`;
            listEl.appendChild(div);
            listEl.scrollTop = listEl.scrollHeight;
        }

        // 手动关闭连接
        // sse.close();
    </script>
</body>

</html>

五、启动

npm run dev

浏览器访问：http://localhost:3000

若端口被占用，先停掉旧进程再启动：

lsof -ti:3000 | xargs kill -9
npm run dev

六、核心原理

1. SSE 消息格式

每条消息必须是：

data: {"type":"delta","id":1,"delta":"服务"}

末尾 两个换行 \n\n 表示一条 event 结束。

2. 自定义协议：delta + done

type 含义 示例
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: "服务" ────────│  id=2 新一轮
  │◄── ping ─────────────────│  10s 心跳

4. 前端打字机逻辑

// 新 id → 新建一行
// 同 id → 往 .text 追加 delta
currentLine.querySelector('.text').textContent += data.delta;

// done → 释放 currentLine，下一轮 delta 会开新行
if (data.type === 'done') currentLine = null;

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 越大，跳字感越强
80 80ms 越小打字越快
5000 5s 两轮回复间隔
10000 10s 心跳间隔

八、和 AI 对话的对应关系

Demo 生产（如灵犀 SSE）
delta 字段 flowResult.outContent 增量
done SSE 连接关闭 + finishReason: "stop"
EventSource GET fetchEventSource POST + 鉴权
textContent += delta markdownRender.addChunk(delta)

这个 Demo 用原生 EventSource + GET，适合理解原理；真实 AI 接口一般是 POST SSE，需要 @microsoft/fetch-event-source 等库。

九、常见问题

Q：看不到打字效果，整句一次出来？
多半是旧版 server 还在跑，重启 node server.js，或在 Network 里确认收到的是 {"type":"delta",...} 而不是整句 JSON。

Q：onerror 一直红字？
EventSource 断线会自动重连，onerror 可能短暂出现，一般可恢复。

Q：想改成 POST SSE？
原生 EventSource 不支持 POST，需换 fetch + 流式读 body，或 @microsoft/fetch-event-source。

十、小结

这个 Demo 覆盖了 SSE 流式 UI 的关键链路：

  1. 服务端：长连接 + res.write 逐段推 delta
  2. 客户端：EventSource.onmessage + 增量追加
  3. 工程化：心跳保活、req.on('close') 清理

三个文件、npm run dev、打开浏览器——就能本地复现 AI 式打字机效果。

# 开发语言
Logo
https://edu.csdn.net/learn/39067/627173?utm_source=2019755004

汇聚全球AI编程工具，助力开发者即刻编程。

更多推荐

cover

当 AI 修图师装上“万能插头“：MCP 如何改变修图工作流

avatar AI编程社区
cover

2026年8月开发者实战指南：ChatGPT Plus / Pro + Codex 如何协同完成需求分析、编码、测试、调试与代码审查

avatar AI编程社区
cover

投资一台B300，部署Claude Opus 5多久才能回本？

avatar AI编程社区