SSE 服务端事件推送系统学习文档

核对时间：2026-07-01
适合人群：前端、Node.js、NestJS、Python FastAPI、Java Spring、AI 应用、前端架构师候选人
学习目标：系统掌握 SSE 的协议格式、浏览器 API、断线重连、服务端实现、Nginx 部署、鉴权、安全、监控和项目落地，并能在 AI 流式聊天、任务进度、日志流、实时通知等场景中独立设计和实现。

SSE 是什么
SSE 和轮询、WebSocket、HTTP Stream 的区别
SSE 协议格式
浏览器 EventSource
前端框架落地：Vue、Nuxt、React
Node.js 与 Express 实现 SSE
NestJS 实现 SSE
FastAPI 实现 SSE
Spring Boot 实现 SSE
项目落地一：AI 流式聊天
项目落地二：任务进度与日志流
项目落地三：实时通知中心
鉴权、安全与权限控制
断线重连、Last-Event-ID 与消息续读
部署：Nginx、网关、CDN 与超时
性能、背压与横向扩展
测试、监控与故障排查
学习路线与练习项目
官方参考资料

1. SSE 是什么

SSE 是 Server-Sent Events，中文常叫“服务端事件推送”。它允许浏览器通过一条 HTTP 长连接持续接收服务端推送的文本事件。

传统 HTTP：

text

浏览器请求 -> 服务端返回完整响应 -> 连接结束

SSE：

text

浏览器建立连接 -> 服务端持续写入事件 -> 浏览器逐条接收 -> 连接保持

它是单向通信：

text

客户端  -> 服务端：建立连接
服务端  -> 客户端：持续推送事件

客户端如果要给服务端发送数据，通常用普通 HTTP：

text

POST /api/chat/messages          提交用户输入
GET  /api/chat/messages/:id/sse  接收模型流式输出

SSE 适合：

AI 聊天逐字输出。
任务进度。
日志实时展示。
导入、导出、构建、部署进度。
站内通知。
审批流状态。
低频或中频运营数据更新。

SSE 不适合：

IM 聊天中客户端和服务端都频繁发消息。
多人协同编辑。
游戏实时同步。
二进制高频传输。
音视频。

这些场景通常更适合 WebSocket 或 WebRTC。

1.1 SSE 不是什么（常见误解澄清）

SSE 不是“协议栈里的新协议”，它就是普通 HTTP 响应不结束：Content-Type: text/event-stream 的、以 chunked 或流式方式持续写入的响应体。TLS、代理、鉴权、路由全部复用 HTTP 生态。
SSE 推的是UTF-8 文本。二进制要先转 Base64（体积 +33% 左右）或拆进多个 data: 字段，不如直接换 WebSocket。
SSE 的连接是浏览器托管的：只要不调 close()，EventSource 掉线后会自动重连，应用代码拿不到“放弃重连”的钩子（除非 close()）。
SSE 事件是至少一次都不保证的：连接断开瞬间正在发送的事件可能丢，所以关键业务必须配 id + 服务端事件表/Redis Streams 续读（见第 14 章），不能假设“写进 res.write 就等于送达”。

💡 一句话记忆：SSE = “一个永远不结束的 GET 响应 + 一套行协议 + 浏览器自动重连”。

2. SSE 和轮询、WebSocket、HTTP Stream 的区别

2.1 与轮询对比

短轮询：

javascript

setInterval(async () => {
  const task = await fetch("/api/tasks/1").then(res => res.json());
  render(task);
}, 3000);

问题：

有延迟。
没有数据也会发请求。
高频轮询会压垮 API 和数据库。

SSE：

javascript

const source = new EventSource("/api/tasks/1/events");
source.addEventListener("progress", (event) => {
  render(JSON.parse(event.data));
});

服务端有事件才推送，延迟更低，请求更少。

2.2 与 WebSocket 对比

对比	SSE	WebSocket
通信方向	服务端到客户端单向推送	双向
协议基础	HTTP	WebSocket 协议
浏览器 API	`EventSource`	`WebSocket`
自动重连	原生支持	需要自己实现
消息格式	文本事件流	文本或二进制
鉴权方式	Cookie、短期 ticket、query token	握手鉴权、auth、首包鉴权
适合	AI 流、进度、日志、通知	聊天、协同、游戏、双向实时

选择原则：

只是服务端持续推前端：优先 SSE。
前端也要高频发消息：WebSocket。
需要二进制：WebSocket。
需要浏览器自动重连：SSE 更省心。

2.3 与 HTTP Stream 对比

HTTP Stream 通常用 fetch().body.getReader() 读取响应流：

javascript

const response = await fetch("/api/chat/stream", { method: "POST" });
const reader = response.body.getReader();

对比	SSE	HTTP Stream
浏览器 API	`EventSource`	`fetch` + `ReadableStream`
请求方法	原生主要 GET	GET/POST 都可
协议格式	`event/data/id/retry`	自定义
自动重连	支持	自己实现
事件类型	原生支持	自己解析
适合	事件推送、任务状态、日志、AI token	直接 POST 后读取模型流、自定义协议

如果你想“POST 一段 prompt，然后直接读取流”，HTTP Stream 更自然。如果你想“先创建任务，再订阅事件”，SSE 更清晰。

2.4 底层差异：一条 SSE、一次 WebSocket 握手、一轮长轮询分别发生了什么

三者的差别不在“能不能实时”，而在链路上反复付出的成本。

长轮询：每一轮都是一次完整的 HTTP 请求-响应周期，靠 Connection: keep-alive 复用 TCP 连接，但 HTTP 头部、请求解析、路由鉴权每轮都要重来一遍：

text

第 1 轮: GET /poll  ----(服务端挂起 25s)----> 200 {msg}   ← 请求头+响应头 各一份
第 2 轮: GET /poll  ----(服务端挂起 25s)----> 200 {msg}   ← 又一份完整 HTTP 事务
第 N 轮: ...

SSE：一次请求头之后，响应体变成一个不结束的字节流，事件只是流里的文本行，没有逐事件 HTTP 开销：

text

GET /events ──> 200 + text/event-stream（响应头仅此一次）
                data: {...}\n\n   ← 事件只是流里的字节
                data: {...}\n\n

WebSocket：先用 HTTP 完成一次 Upgrade 握手拿到 101 Switching Protocols，之后这条 TCP 连接上跑的就不再是 HTTP，而是 WebSocket 帧（RFC 6455）：

text

GET /ws  Upgrade: websocket  →  101 Switching Protocols   ← 唯一一次 HTTP
[帧][帧][帧]...  帧头仅 2~10 字节，双向；客户端→服务端帧必须 mask

每条消息的链路开销大致对比：

方式	每条消息附加开销	方向	连接形态
短/长轮询	一个完整 HTTP 事务（头部可达数百字节~KB 级）	只能客户端发起	多次短连接复用
SSE	只有 `event:/data:` 字段文本本身	服务端 → 客户端	一条不结束的 HTTP 响应
WebSocket	帧头 2~14 字节（取决于长度编码；客户端帧另加 4 字节 mask key，小消息典型 6 字节）	双向	一条升级后的 TCP 连接

由此推出三个工程结论：

高频小消息时轮询的头部开销被放大最严重，WebSocket 最省，SSE 居中（文本字段本身也是开销）。
代理与鉴权链路：SSE 全程是 HTTP，网关/WAF/日志几乎零改造；WebSocket 需要网关透传 Upgrade/Connection 头，否则握手失败。
浏览器连接配额：HTTP/1.1 下浏览器对同一域名的并发连接数有限（经典值为 6，Chrome/Firefox 对 SSE 挤占该配额长期标记为 Won't Fix）；开多个标签页的 SSE 会互相争抢。HTTP/2 下 SSE 走多路复用的 stream（并发上限由协商决定，默认常见 100），不再占 6 连接配额——这是“SSE 多标签页卡死”问题治本的方式之一。

3. SSE 协议格式

3.1 响应头

服务端必须返回：

http

Content-Type: text/event-stream
Cache-Control: no-cache
Connection: keep-alive

常见增强：

http

X-Accel-Buffering: no

用于提示 Nginx 不要缓冲响应。

3.2 基础事件

text

data: hello

每条事件以空行结束。

浏览器收到：

javascript

source.onmessage = (event) => {
  console.log(event.data); // hello
};

3.3 自定义事件名

text

event: progress
data: {"percent":50,"message":"处理中"}

event: done
data: {"status":"success"}

前端监听：

javascript

source.addEventListener("progress", (event) => {
  const data = JSON.parse(event.data);
  console.log(data.percent);
});

source.addEventListener("done", (event) => {
  console.log(JSON.parse(event.data));
  source.close();
});

3.4 id 与断线续读

text

id: 1001
event: progress
data: {"percent":50}

浏览器断线重连时会带上：

http

Last-Event-ID: 1001

服务端可以从 1001 之后继续推，避免丢事件。

3.5 retry

text

retry: 3000
data: connected

表示建议浏览器断线后 3000ms 再重连。

3.6 多行 data

text

data: 第一行
data: 第二行

浏览器会把多行 data 合并，中间带换行。

3.7 心跳

SSE 可以发送注释行保持连接：

text

: heartbeat

浏览器忽略注释，但代理和连接会看到有数据流动。

3.8 内部机制：浏览器如何逐行解析事件流

事件流的解析规则由 HTML 标准（WHATWG §9.2.4 The event stream format）定义，理解它对排查“事件不触发”类问题非常关键：

字节 → 字符：流必须按 UTF-8 解码，开头的 BOM（U+FEFF）若存在会被剥离。字节不完整时浏览器会等待凑齐再解码，所以多字节字符跨 chunk 不会被切碎。
行分隔符：\r\n、\r、\n 都算合法行尾（标准 ABNF：end-of-line = cr lf / cr / lf），空行表示“派发一个事件”。
字段切分：每行按第一个冒号 : 切成字段名和值；值开头的一个空格会被剥掉。所以 data:x 与 data: x 等价，而 data: x（两个空格）的值开头保留一个空格。
无冒号的行：整行是字段名、值为空字符串（例如单独一行 data 表示追加一个空行）。
开头的冒号：整行是注释（: heartbeat），被忽略——这就是心跳的协议依据。
未知字段直接忽略：标准只定义 data/event/id/retry，自己扩展字段不会报错，但也不要指望旧浏览器理解。
字段累积：同一事件内多个 data: 行用 \n 连接；event/id 取该事件内最后一次出现的值；空行派发后缓冲区重置（last event ID 除外，它跨事件保留）。

text

字节流                                   解析结果
data: {"a":1}      ─┐
                    ├ 空行 → dispatch: MessageEvent{ data:'{"a":1}' }
                    │
event: ping        ─┐
data: 1            ├ 空行 → dispatch: ping{ data:'1' }
                    │
: heartbeat        ──── 注释，忽略

⚠️ 由此产生的经典坑：

data 里不能含裸 \n：JSON 字符串里换行必须是 \\n 转义，否则一个 \n 就提前“结束 data 行”，把 payload 撕碎——JSON.stringify 天然会转义，所以“data 永远放 JSON”是安全的。
CR/LF 混用没问题，但 id 字段值不能含 NUL（U+0000），含 NUL 的 id 会被标准直接忽略。
服务端只写 \n 不写 \n\n：事件永远不会派发，表现为“curl 能看到字节、浏览器收不到事件”——排查时先看响应里有没有成对换行。

3.9 内部机制：`Connection: keep-alive` 在 SSE 里其实是摆设

注意第 3.1 节的响应头三件套里，Connection: keep-alive 只对 HTTP/1.1 有历史意义（HTTP/1.1 默认持久连接；HTTP/2、HTTP/3 禁止 Connection 这类逐跳头，Node 的 HTTP/2 实现会直接丢弃它）。SSE 真正依赖的是响应体不结束，而不是 keep-alive 本身。写它只是为了兼容某些老代理的猜测式行为，删掉一般也不影响功能；Content-Type: text/event-stream 和防止缓冲才是生死线。

💡 HTTP/2 下若响应迟迟不到，优先怀疑中间层缓冲和 proxy_read_timeout，而不是连接复用参数。

4. 浏览器 EventSource

4.1 基础用法

javascript

const source = new EventSource("/api/events");

source.onopen = () => {
  console.log("SSE connected");
};

source.onmessage = (event) => {
  console.log(event.data);
};

source.onerror = (event) => {
  console.log("SSE error", event);
};

4.2 关闭连接

javascript

source.close();

如果组件卸载时不关闭连接，会造成：

无用网络连接。
服务端连接不释放。
重复收到消息。
内存泄漏。

同域请求会自动带 Cookie。

跨域且需要 Cookie：

javascript

const source = new EventSource("https://api.example.com/events", {
  withCredentials: true,
});

服务端需要允许 credentials：

http

Access-Control-Allow-Origin: https://app.example.com
Access-Control-Allow-Credentials: true

不能使用 * 搭配 credentials。

4.4 EventSource 的限制

原生 EventSource 不能自定义请求头。也就是说不能这样：

javascript

new EventSource("/events", {
  headers: {
    Authorization: "Bearer token",
  },
});

常见替代方案：

使用 Cookie session。
先用普通 HTTP 换取短期 ticket，再用 query 传 ticket。
使用支持 fetch 的 SSE polyfill。
改用 HTTP Stream，用 fetch 自己读流。

短期 ticket：

text

POST /api/realtime-ticket
GET /api/tasks/1/events?ticket=rt_abc

ticket 要短期有效、一次性或可撤销，不能长期暴露。

4.5 内部机制：EventSource 的状态机与自动重连规则

浏览器处理 SSE 连接的模型定义在 HTML 标准 §9.2.3（Processing model），把几条关键分支记住，能解释大部分“为什么重连了/为什么不再重连了”的现象：

text

new EventSource(url)
   │  fetch（cache mode 固定为 no-store，浏览器不会用 HTTP 缓存敷衍你）
   ▼
响应到达 ──网络错误?─────────────── 是 → reestablish（自动重连）
   │否
   ├──状态码 ≠ 200 或 Content-Type ≠ text/event-stream
   │        └─ 是 → fail the connection：readyState=CLOSED，只发一次 error，
   │                浏览器不再重连（服务端想"劝退"就用 204 No Content）
   │否
   ▼
announce the connection（发 open 事件，readyState=OPEN）→ 逐行解析并派发事件
   │
   └──流结束/连接断开 → reestablish（自动重连）

重连的细节：

重连开始时先派发一次 error 事件、readyState 回到 CONNECTING（0）——所以 onerror 不等于“彻底失败”，它也可能只是“正在等下一次重连”。
等待时长 = reconnection time：初始值是浏览器实现自定义（通常是几秒），服务端可用 retry: 字段覆盖（只接受纯 ASCII 数字，单位 ms）。标准还允许浏览器在反复失败时自行加指数退避。
只要 last event ID string 非空，重连请求就会带 Last-Event-ID 头（UTF-8 编码，不得含 NUL/LF/CR）。
页面关闭/文档销毁时浏览器强制关闭连接；EventSource 对象被 GC 时连接也会被中断——但别指望 GC 来管理连接，仍应显式 close()。

⚠️ 一个常被误诊的行为：fail the connection（状态码非 200，或 Content-Type 不是 text/event-stream）后浏览器不会自动重试，onerror 后 readyState 停在 CLOSED。网络层错误（含 CORS 被拒产生的 network error）则走 reestablish 自动重连，直到浏览器“知道重试无望”才放弃。排查“怎么不重连了”先看第一条分支。

4.6 注意事项：`open` 与首事件、Worker 中的 EventSource

open 事件在响应头被确认为合法流时触发，不保证已收到任何数据；不要拿 onopen 当“收到 connected 事件”用。
event: open 这种自定义事件名不与原生 open 事件冲突（原生 open 不是 MessageEvent），但 message/error/open 是保留的默认派发名，谨慎覆盖语义。
EventSource 在 Web Worker 中可用（MDN 明确标注 available in Web Workers），适合“后台常驻通知连接 + BroadcastChannel 分发给各标签页”，顺便解决多标签页重复连接问题。
每个事件都携带 MessageEvent.lastEventId 属性，取的是流中最后一次 id: 的值（即使本事件没有 id 字段，也会沿用上一个 id），做去重和游标时直接用这个属性即可。

5. 前端框架落地：Vue、Nuxt、React

5.1 Vue composable：useSse

typescript

import { onBeforeUnmount, ref } from "vue";

export function useSse<T>(url: string) {
  const connected = ref(false);
  const error = ref<Event | null>(null);
  const lastEvent = ref<T | null>(null);

  let source: EventSource | null = null;

  function connect() {
    source = new EventSource(url, {
      withCredentials: true,
    });

    source.onopen = () => {
      connected.value = true;
      error.value = null;
    };

    source.onmessage = (event) => {
      lastEvent.value = JSON.parse(event.data);
    };

    source.onerror = (event) => {
      connected.value = false;
      error.value = event;
    };
  }

  function listen<K>(eventName: string, handler: (data: K) => void) {
    source?.addEventListener(eventName, (event) => {
      handler(JSON.parse((event as MessageEvent).data));
    });
  }

  function close() {
    source?.close();
    source = null;
    connected.value = false;
  }

  connect();
  onBeforeUnmount(close);

  return {
    connected,
    error,
    lastEvent,
    listen,
    close,
  };
}

任务进度组件：

vue

<script setup lang="ts">
import { ref } from "vue";
import { useSse } from "@/composables/useSse";

const props = defineProps<{
  taskId: string;
}>();

const percent = ref(0);
const message = ref("");

const { listen, connected, close } = useSse(`/api/tasks/${props.taskId}/events`);

listen<{ percent: number; message: string }>("progress", (data) => {
  percent.value = data.percent;
  message.value = data.message;
});

listen("done", () => {
  close();
});
</script>

<template>
  <section>
    <p>{{ connected ? "连接中" : "已断开" }}</p>
    <progress :value="percent" max="100" />
    <p>{{ message }}</p>
  </section>
</template>

5.2 Vue + Pinia：通知中心

typescript

import { defineStore } from "pinia";

export const useNotificationStore = defineStore("notification", {
  state: () => ({
    unreadCount: 0,
    items: [] as Array<{ id: string; title: string; createdAt: string }>,
  }),
  actions: {
    add(item: { id: string; title: string; createdAt: string }) {
      if (this.items.some(existing => existing.id === item.id)) return;
      this.items.unshift(item);
      this.unreadCount += 1;
    },
  },
});

接入 SSE：

typescript

const notificationStore = useNotificationStore();
const source = new EventSource("/api/notifications/events", {
  withCredentials: true,
});

source.addEventListener("notification", (event) => {
  notificationStore.add(JSON.parse((event as MessageEvent).data));
});

5.3 Nuxt 注意点

Nuxt 有 SSR，不能在服务端渲染阶段直接使用 EventSource。

vue

<script setup lang="ts">
const messages = ref<string[]>([]);

onMounted(() => {
  const source = new EventSource("/api/chat/1/events");

  source.addEventListener("token", (event) => {
    messages.value.push(JSON.parse((event as MessageEvent).data).text);
  });

  onBeforeUnmount(() => {
    source.close();
  });
});
</script>

规则：

EventSource 放在 onMounted 或客户端插件里。
服务端路由可作为 BFF 转发上游 SSE。
不要在 SSR 阶段读取 window。

5.4 React hook

tsx

import { useEffect, useState } from "react";

export function useSse<T>(url: string, eventName = "message") {
  const [data, setData] = useState<T | null>(null);
  const [connected, setConnected] = useState(false);

  useEffect(() => {
    const source = new EventSource(url, {
      withCredentials: true,
    });

    source.onopen = () => setConnected(true);
    source.onerror = () => setConnected(false);

    source.addEventListener(eventName, (event) => {
      setData(JSON.parse((event as MessageEvent).data));
    });

    return () => {
      source.close();
    };
  }, [url, eventName]);

  return { data, connected };
}

6. Node.js 与 Express 实现 SSE

6.1 最小 SSE 接口

javascript

import express from "express";

const app = express();

app.get("/api/events", (req, res) => {
  res.setHeader("Content-Type", "text/event-stream; charset=utf-8");
  res.setHeader("Cache-Control", "no-cache, no-transform");
  res.setHeader("Connection", "keep-alive");
  res.setHeader("X-Accel-Buffering", "no");
  res.flushHeaders?.();

  res.write("retry: 3000\n");
  res.write(`data: ${JSON.stringify({ message: "connected" })}\n\n`);

  const timer = setInterval(() => {
    res.write(`event: time\n`);
    res.write(`data: ${JSON.stringify({ now: new Date().toISOString() })}\n\n`);
  }, 1000);

  req.on("close", () => {
    clearInterval(timer);
  });
});

6.2 封装 sendEvent

javascript

function sendEvent(res, event, data, id) {
  if (id) {
    res.write(`id: ${id}\n`);
  }
  if (event) {
    res.write(`event: ${event}\n`);
  }
  res.write(`data: ${JSON.stringify(data)}\n\n`);
}

使用：

javascript

sendEvent(res, "progress", {
  percent: 50,
  message: "处理中",
}, "1001");

6.3 任务进度接口

javascript

app.get("/api/tasks/:taskId/events", async (req, res) => {
  const { taskId } = req.params;

  res.setHeader("Content-Type", "text/event-stream; charset=utf-8");
  res.setHeader("Cache-Control", "no-cache, no-transform");
  res.setHeader("Connection", "keep-alive");
  res.setHeader("X-Accel-Buffering", "no");
  res.flushHeaders?.();

  const timer = setInterval(async () => {
    const task = await taskService.getTask(taskId);

    sendEvent(res, "progress", {
      percent: task.percent,
      message: task.message,
      status: task.status,
    });

    if (task.status === "success" || task.status === "failed") {
      sendEvent(res, "done", task);
      clearInterval(timer);
      res.end();
    }
  }, 1000);

  req.on("close", () => {
    clearInterval(timer);
  });
});

注意：

不要每秒查数据库给大量连接用；高并发时应改成订阅 Redis Streams、Pub/Sub 或任务内存事件。
每个连接断开后必须清理 timer 和 listener。
res.write 返回 false 时说明缓冲区压力较大，要考虑背压和慢客户端。

6.4 Node 原生 HTTP

javascript

import http from "node:http";

http.createServer((req, res) => {
  if (req.url !== "/events") {
    res.writeHead(404);
    res.end();
    return;
  }

  res.writeHead(200, {
    "Content-Type": "text/event-stream; charset=utf-8",
    "Cache-Control": "no-cache, no-transform",
    "Connection": "keep-alive",
  });

  const timer = setInterval(() => {
    res.write(`data: ${Date.now()}\n\n`);
  }, 1000);

  req.on("close", () => {
    clearInterval(timer);
  });
}).listen(3000);

7. NestJS 实现 SSE

7.1 @Sse 基础

NestJS 的 @Sse() 通常返回 RxJS Observable。

typescript

import { Controller, MessageEvent, Sse } from "@nestjs/common";
import { interval, map, Observable } from "rxjs";

@Controller("events")
export class EventsController {
  @Sse()
  stream(): Observable<MessageEvent> {
    return interval(1000).pipe(
      map((index) => ({
        type: "time",
        data: {
          index,
          now: new Date().toISOString(),
        },
      })),
    );
  }
}

7.2 任务进度

typescript

import { Controller, MessageEvent, Param, Sse } from "@nestjs/common";
import { interval, map, Observable, takeWhile } from "rxjs";

@Controller("tasks")
export class TaskEventsController {
  constructor(private readonly taskService: TaskService) {}

  @Sse(":taskId/events")
  events(@Param("taskId") taskId: string): Observable<MessageEvent> {
    return interval(1000).pipe(
      map(() => this.taskService.getProgress(taskId)),
      takeWhile((task) => task.status === "running", true),
      map((task) => ({
        type: task.status === "running" ? "progress" : "done",
        data: task,
      })),
    );
  }
}

7.3 从 Subject 推送

适合由任务服务主动推送进度：

typescript

import { Injectable, MessageEvent } from "@nestjs/common";
import { Observable, Subject } from "rxjs";

@Injectable()
export class TaskEventService {
  private readonly subjects = new Map<string, Subject<MessageEvent>>();

  stream(taskId: string): Observable<MessageEvent> {
    const subject = this.subjects.get(taskId) ?? new Subject<MessageEvent>();
    this.subjects.set(taskId, subject);
    return subject.asObservable();
  }

  emit(taskId: string, event: string, data: unknown) {
    const subject = this.subjects.get(taskId);
    subject?.next({ type: event, data });
  }

  complete(taskId: string) {
    const subject = this.subjects.get(taskId);
    subject?.complete();
    this.subjects.delete(taskId);
  }
}

Controller：

typescript

@Sse(":taskId/events")
events(@Param("taskId") taskId: string) {
  return this.taskEventService.stream(taskId);
}

生产注意：内存 Map 只适合单实例。多实例时要接 Redis Streams、Redis Pub/Sub、Kafka 或数据库事件表。

8. FastAPI 实现 SSE

8.1 StreamingResponse

python

import asyncio
import json
from fastapi import FastAPI
from fastapi.responses import StreamingResponse

app = FastAPI()

async def event_generator():
    index = 0
    while True:
        await asyncio.sleep(1)
        data = json.dumps({"index": index}, ensure_ascii=False)
        yield f"event: time\ndata: {data}\n\n"
        index += 1

@app.get("/api/events")
async def events():
    return StreamingResponse(
        event_generator(),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "X-Accel-Buffering": "no",
        },
    )

8.2 任务进度

python

async def task_events(task_id: str):
    while True:
        task = await get_task(task_id)
        event = "done" if task["status"] in ["success", "failed"] else "progress"
        yield f"event: {event}\ndata: {json.dumps(task, ensure_ascii=False)}\n\n"

        if event == "done":
            break

        await asyncio.sleep(1)

@app.get("/api/tasks/{task_id}/events")
async def task_event_stream(task_id: str):
    return StreamingResponse(
        task_events(task_id),
        media_type="text/event-stream",
    )

8.3 队列驱动

python

task_queues: dict[str, asyncio.Queue[dict]] = {}

async def publish_progress(task_id: str, event: dict):
    queue = task_queues.setdefault(task_id, asyncio.Queue())
    await queue.put(event)

async def sse_from_queue(task_id: str):
    queue = task_queues.setdefault(task_id, asyncio.Queue())

    while True:
        event = await queue.get()
        event_name = event["event"]
        yield f"event: {event_name}\ndata: {json.dumps(event, ensure_ascii=False)}\n\n"

        if event_name == "done":
            break

生产环境应把内存队列换成 Redis Streams、Kafka、数据库状态表或任务系统。

9. Spring Boot 实现 SSE

9.1 Spring MVC SseEmitter

java

@RestController
@RequestMapping("/api/tasks")
public class TaskSseController {
    private final TaskService taskService;

    public TaskSseController(TaskService taskService) {
        this.taskService = taskService;
    }

    @GetMapping(value = "/{taskId}/events", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public SseEmitter events(@PathVariable String taskId) {
        SseEmitter emitter = new SseEmitter(0L);

        taskService.watch(taskId, progress -> {
            try {
                emitter.send(SseEmitter.event()
                    .name(progress.finished() ? "done" : "progress")
                    .data(progress));

                if (progress.finished()) {
                    emitter.complete();
                }
            } catch (IOException e) {
                emitter.completeWithError(e);
            }
        });

        emitter.onCompletion(() -> taskService.unwatch(taskId));
        emitter.onTimeout(() -> taskService.unwatch(taskId));
        emitter.onError(error -> taskService.unwatch(taskId));

        return emitter;
    }
}

适合传统 Spring MVC 项目。

9.2 Spring WebFlux

java

@RestController
@RequestMapping("/api/tasks")
public class TaskFluxController {
    private final TaskService taskService;

    public TaskFluxController(TaskService taskService) {
        this.taskService = taskService;
    }

    @GetMapping(value = "/{taskId}/events", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public Flux<ServerSentEvent<TaskProgress>> events(@PathVariable String taskId) {
        return taskService.progressFlux(taskId)
            .map(progress -> ServerSentEvent.<TaskProgress>builder()
                .event(progress.finished() ? "done" : "progress")
                .data(progress)
                .build());
    }
}

WebFlux 适合高并发连接和响应式链路。不要在 WebFlux 中直接执行阻塞数据库或阻塞 SDK 调用；必须调用时要隔离到合适线程池。

10. 项目落地一：AI 流式聊天

10.1 需求

实现类似 ChatGPT 的流式回答：

用户提交问题。
服务端调用模型。
前端逐字显示回答。
支持停止生成。
生成完成后保存完整消息。
记录 token、耗时、错误。

10.2 推荐架构

text

Vue / Nuxt
  -> POST /api/chat/messages
  -> GET /api/chat/messages/:messageId/events

Node/Nest/FastAPI/Spring BFF
  -> 校验用户和会话权限
  -> 调用模型服务流式接口
  -> 转成 SSE token/done/error 事件
  -> 保存完整消息和调用日志

Database
  -> conversations
  -> messages
  -> model_call_logs

10.3 接口设计

创建消息：

http

POST /api/chat/conversations/:conversationId/messages
Content-Type: application/json

{
  "content": "解释一下 SSE 的项目落地方式"
}

json

{
  "messageId": "msg_123",
  "streamUrl": "/api/chat/messages/msg_123/events"
}

http

GET /api/chat/messages/msg_123/events
Accept: text/event-stream

事件：

text

event: token
data: {"text":"SSE"}

event: token
data: {"text":" 适合"}

event: done
data: {"messageId":"msg_123","usage":{"inputTokens":100,"outputTokens":300}}

10.4 前端处理

typescript

async function sendMessage(conversationId: string, content: string) {
  const response = await fetch(`/api/chat/conversations/${conversationId}/messages`, {
    method: "POST",
    headers: { "Content-Type": "application/json" },
    body: JSON.stringify({ content }),
  });

  const { messageId, streamUrl } = await response.json();
  const assistantMessage = {
    id: messageId,
    role: "assistant",
    content: "",
    status: "streaming",
  };

  const source = new EventSource(streamUrl, {
    withCredentials: true,
  });

  source.addEventListener("token", (event) => {
    const { text } = JSON.parse((event as MessageEvent).data);
    assistantMessage.content += text;
  });

  source.addEventListener("done", () => {
    assistantMessage.status = "done";
    source.close();
  });

  source.addEventListener("error", () => {
    assistantMessage.status = "error";
    source.close();
  });
}

10.5 后端关键设计

必须考虑：

浏览器断开后是否继续生成。
用户点击停止生成如何取消上游模型请求。
流式内容是边生成边保存，还是最终保存。
如何处理模型异常、超时、内容安全。
如何限制单用户并发生成数。
如何统计 token 和费用。
如何防止用户订阅不属于自己的 messageId。

停止生成：

http

DELETE /api/chat/messages/:messageId/stream

服务端用 AbortController、任务状态或框架取消机制终止上游流。

11. 项目落地二：任务进度与日志流

11.1 需求

后台导入 Excel：

上传文件。
后端异步解析。
前端实时看到进度。
错误行实时显示。
完成后下载错误报告。

11.2 架构

text

Vue
  -> POST /api/imports 上传文件
  -> 返回 taskId
  -> GET /api/imports/:taskId/events 订阅 SSE

API Server
  -> 创建任务记录
  -> 投递任务到队列

Worker
  -> 分批处理
  -> 写进度事件到 Redis Streams / DB

SSE Endpoint
  -> 读取进度事件
  -> 推送给前端

11.3 事件设计

text

event: progress
data: {"stage":"validate","percent":40,"message":"已校验 4000 / 10000 行"}

event: row_error
data: {"row":1024,"message":"手机号格式错误"}

event: done
data: {"successCount":9900,"failedCount":100,"errorReportUrl":"/api/imports/task_1/errors.xlsx"}

11.4 为什么适合 SSE

任务进度是服务端单向告诉前端：

前端不需要频繁发消息。
HTTP 上传和下载仍然走普通接口。
SSE 专门负责进度事件。
服务端和前端实现都比 WebSocket 简单。

12. 项目落地三：实时通知中心

12.1 需求

管理后台需要：

新订单通知。
审批待办提醒。
风险告警。
顶部未读数量。
多业务服务都能发布通知。

12.2 架构

text

Order Service / Risk Service / Workflow Service
  -> Redis Streams / Kafka / RabbitMQ
Notification Service
  -> 持久化通知
  -> SSE 推送给在线用户
Vue Admin
  -> Pinia notification store

12.3 SSE 连接

javascript

const source = new EventSource("/api/notifications/events", {
  withCredentials: true,
});

source.addEventListener("notification", (event) => {
  notificationStore.add(JSON.parse(event.data));
});

source.addEventListener("unread_count", (event) => {
  notificationStore.setUnreadCount(JSON.parse(event.data).count);
});

12.4 后端发布事件

text

event: notification
data: {"id":"n_1","title":"新订单","createdAt":"2026-07-01T10:00:00Z"}

event: unread_count
data: {"count":12}

12.5 多实例问题

如果 API 服务有多个实例，不能只把连接存在本机内存后直接推。

13. 鉴权、安全与权限控制

同域 Cookie 最简单：

javascript

const source = new EventSource("/api/tasks/1/events");

服务端从 session/JWT cookie 中识别用户。

优点：

EventSource 原生支持。
不暴露 token 到 URL。

注意：

配置 SameSite、Secure、HttpOnly。
跨域时设置 CORS credentials。

13.2 Ticket 鉴权

如果系统不用 Cookie，可先换短期 ticket：

text

POST /api/realtime-ticket
GET /api/tasks/1/events?ticket=rt_abc

ticket 规则：

短有效期，例如 30 秒。
绑定 userId、taskId、用途。
可一次性使用。
服务端校验后立即作废或标记。

13.3 权限校验

SSE 连接不能只校验“已登录”，还要校验资源权限：

text

用户 A 是否能订阅 task_123？
用户 A 是否属于 conversation_456？
用户 A 是否能接收 tenant_1 的通知？

连接建立前校验：

text

GET /api/tasks/:taskId/events
  -> auth user
  -> check task owner / tenant / permission
  -> allow stream

13.4 安全注意

不要把长期 token 放在 URL。
不要允许用户订阅任意 taskId/messageId。
控制每个用户最大 SSE 连接数。
控制消息大小。
日志不要记录敏感 query。
对事件数据做脱敏。
跨域必须精确配置 origin。

13.5 怎么用 fetch 读 SSE：解开自定义头的限制

原生 EventSource 不能设 Authorization 头，但协议本身没有这个限制——text/event-stream 就是普通响应流，用 fetch + ReadableStream 一样能读，还能 POST、带任意头、用 AbortController 取消（这也是各家 AI SDK 流式客户端的常见做法）：

typescript

export async function* readSse(url: string, init: RequestInit, signal?: AbortSignal) {
  const res = await fetch(url, {
    ...init,
    headers: { ...init.headers, Accept: "text/event-stream" },
    signal,
  });
  if (!res.ok || !res.body) throw new Error(`SSE failed: ${res.status}`);

  const reader = res.body.pipeThrough(new TextDecoderStream()).getReader();
  let buf = "";

  while (true) {
    const { done, value } = await reader.read();
    if (done) break;
    buf += value;

    // 按空行切事件；跨 chunk 的不完整事件留在 buf 里等下一段
    let m: RegExpExecArray | null;
    const SEP = /\r\n\r\n|\n\n|\r\r/g;
    while ((m = SEP.exec(buf)) !== null) {
      const raw = buf.slice(0, m.index);
      buf = buf.slice(m.index + m[0].length);
      SEP.lastIndex = 0;

      const evt = { event: "message", data: "", id: "" };
      const dataLines: string[] = [];
      for (const line of raw.split(/\r\n|\n|\r/)) {
        if (line.startsWith(":")) continue;              // 注释行
        const i = line.indexOf(":");                     // 只看第一个冒号
        const field = i === -1 ? line : line.slice(0, i);
        let val = i === -1 ? "" : line.slice(i + 1);
        if (val.startsWith(" ")) val = val.slice(1);     // 剥掉一个空格
        if (field === "event") evt.event = val;
        else if (field === "data") dataLines.push(val);
        else if (field === "id") evt.id = val;
      }
      if (dataLines.length === 0) continue;              // 纯注释/纯 id 块不派发
      evt.data = dataLines.join("\n");
      yield evt;
    }
  }
}

注意与 EventSource 的行为差异：

没有自动重连：断开后要么自己带 Last-Event-ID 头重发请求，要么换回 EventSource。
解析规则必须自己实现完整（首个冒号、剥一个空格、多行 data 用 \n 拼接、空行派发、跨 chunk 残留缓冲），上面代码是逐条对照规范写的最小实现。
好处：可以 AbortController 精确取消、可以 POST body、事件处理函数不再受 addEventListener 名字限制。

社区成熟实现（@microsoft/fetch-event-source、eventsource-parser 等）已把这些边角处理完，生产建议直接用库而非手写解析。

14. 断线重连、Last-Event-ID 与消息续读

14.1 浏览器自动重连

SSE 断开后，浏览器会自动尝试重连。服务端可用 retry 建议重连间隔：

text

retry: 3000
data: connected

14.2 Last-Event-ID

服务端发送事件 ID：

text

id: 101
event: progress
data: {"percent":50}

断线后浏览器重连时会带：

http

Last-Event-ID: 101

服务端读取：

javascript

app.get("/api/tasks/:taskId/events", (req, res) => {
  const lastEventId = req.header("Last-Event-ID");
});

14.3 续读设计

要支持续读，就不能只把事件写到内存里就忘掉。可以用：

Redis Streams。
Kafka。
数据库 event 表。
任务进度快照 + 最新日志 offset。

Redis Streams 思路：

text

Worker:
XADD task:123:events * event progress percent 50

SSE:
读取 Last-Event-ID
XREAD BLOCK 30000 STREAMS task:123:events lastId

14.4 前端去重

重连可能导致重复事件，前端也要按 id 去重。

typescript

const seen = new Set<string>();

source.addEventListener("progress", (event) => {
  const messageEvent = event as MessageEvent;
  if (seen.has(messageEvent.lastEventId)) return;
  seen.add(messageEvent.lastEventId);

  render(JSON.parse(messageEvent.data));
});

14.5 内部机制：续读闭环的服务端设计

“带上了 Last-Event-ID”只是续读的一半，完整闭环要求服务端把它当消费位点用。典型实现（以 Redis Streams 为例，其 entry ID <ms>-<seq> 天然单调递增，可直接当 SSE 的 id）：

javascript

app.get("/api/tasks/:taskId/events", async (req, res) => {
  openSseHeaders(res);

  // Last-Event-ID 是用户可控输入：必须校验格式，非法就当作无游标
  const lastId = validateStreamId(req.header("Last-Event-ID")) ?? "0-0";

  // 1. 先补发：读取游标之后的历史事件（追 backlog）
  const backlog = await redis.xread("COUNT", 500, "STREAMS", streamKey, lastId);
  for (const [id, fields] of entriesOf(backlog)) writeEvent(res, id, fields);

  // 2. 再阻塞跟读：新事件到达即推送（XREAD BLOCK 让位点精确续接）
  (async () => {
    let cursor = lastReadIdFrom(backlog) ?? lastId;
    while (!res.writableEnded) {
      const batch = await redis.xread("BLOCK", 15000, "COUNT", 100,
                                      "STREAMS", streamKey, cursor);
      if (!batch) { res.write(": heartbeat\n\n"); continue; } // 读超时→心跳
      for (const [id, fields] of entriesOf(batch)) {
        cursor = id;
        writeEvent(res, id, fields);
      }
    }
  })().catch(() => res.end());
});

必须想清楚的边界：

问题	处理原则
事件流已被裁剪（Redis `MAXLEN` / Kafka retention）	无法从该位点续读：返回一个 `event: resume_reset` 让前端回查快照接口，别静默丢
进度类事件重复补发导致回退显示	事件带 `seq`/`percent`，前端按“只接受更大值”合并（幂等渲染）
`Last-Event-ID` 伪造（订阅别人的 task）	位点必须校验属于该 task 的 stream key，权限先于续读
断线期间任务已 `done`	补发 backlog 后直接发 done 并 `res.end()`，不要悬挂
浏览器永不发 `Last-Event-ID`（首次连接）	等价于从“现在”开始（Redis 用 `$`），或按需从头补

💡 SSE 规范只保证“浏览器会把最后 id 带回来”，不保证不丢事件也不保证不重：id + 持久化事件源 = 至少一次投递，前端按 id 去重 + 单调合并 = 实践上的“有效一次”。

另外，HTML 标准 Authoring notes 特别提到：为防老代理短空闲掐连接，建议约每 15 秒发一行注释心跳；还提醒 chunked 编码由不感知时序的中间层重分块可能造成事件延迟甚至乱序——遇到“偶发事件粘连/延迟”，可在网关上对该路由关闭再分块。

15. 部署：Nginx、网关、CDN 与超时

15.1 Nginx 配置

SSE 最大的线上坑是代理缓冲。Nginx 如果缓冲响应，前端会等很久才收到数据。

nginx

location /api/ {
  proxy_pass http://api_upstream;
  proxy_http_version 1.1;
  proxy_set_header Connection "";
  proxy_buffering off;
  proxy_cache off;
  proxy_read_timeout 3600s;
  proxy_send_timeout 3600s;
}

服务端也建议返回：

http

X-Accel-Buffering: no
Cache-Control: no-cache, no-transform

15.2 网关和 CDN

注意：

某些 API Gateway 默认会缓冲响应。
某些 CDN 不适合代理 SSE。
中间层可能有 60s/100s/300s idle timeout。
长时间无消息时要发送心跳。

心跳：

text

: heartbeat

15.3 HTTP/1.1 连接限制

浏览器在 HTTP/1.1 下对同域并发连接数有限制。多个标签页都打开 SSE 时可能挤占连接。

优化：

尽量复用一个全局 SSE 连接。
用 HTTP/2。
不要每个组件都创建独立 EventSource。
后台页面关闭时主动 close。

15.4 内部机制：一条 SSE 上到底叠了几层缓冲

“本地实时、线上不实时”的根因，几乎都藏在下面这条链路的某一层缓冲里。数据要穿过这些层才到 JS：

text

res.write()
  ① Node 内部写缓冲（highWaterMark，默认 16KiB）
  ② TCP 发送缓冲 + Nagle 算法（可能合并小包）
  ③ Nginx: proxy_buffering（默认 on，按 proxy_buffer 大小攒块）
  ④ Nginx: gzip / sub_filter（攒够一块才压，破坏流式）
  ⑤ 云网关 / CDN（各自有缓冲与 idle timeout）
  ⑥ 浏览器流解码 → 事件解析 → 任务队列派发

逐层的关闭手段：

层	症状	处理
③ proxy_buffering	事件成批到达、间隔很久	`proxy_buffering off`，或服务端响应头 `X-Accel-Buffering: no`（Nginx 专有，让后端自行关闭本响应缓冲）
④ gzip	前几 KB 正常后卡住、结尾一次性吐出	对 `text/event-stream` 关闭压缩：`gzip_types` 不含它，并回 `Cache-Control: no-transform`
② Nagle	极小事件（几十字节）延迟几十 ms	Node 上 `res.socket?.setNoDelay(true)`
① Node 写缓冲	内存涨、`write` 返回 false	按 `drain` 退避（见 16.5）
⑤ 网关/CDN	固定时长被切、跨区首包慢	查该产品的 SSE/streaming 支持与 idle timeout；不可控就让该域名不走 CDN

💡 定位顺序建议：先 curl -N 直连源站看是否实时 → 再 curl -N 走网关 → 差异点就是出问题的那一层。这一步能省掉半小时瞎改配置。

⚠️ 不要用 Cache-Control: no-store 来“防缓冲”。no-store 管的是要不要把响应存进 HTTP 缓存，与代理的传输期缓冲是两回事；关掉缓冲靠 X-Accel-Buffering: no / proxy_buffering off，缓存侧写 no-cache, no-transform 即可（SSE 请求本身的 cache mode 已是 no-store，见 4.5）。

16. 性能、背压与横向扩展

16.1 连接数量

每个 SSE 连接都是一条长连接，会占用：

文件描述符。
内存。
网关连接。
服务端连接上下文。

服务端要限制：

单用户连接数。
单 IP 连接数。
单租户连接数。
空闲连接时间。

16.2 发送频率

不要把高频指标逐条推给前端。可以合并：

text

每 100ms 聚合一次指标
每 1s 推一次看板快照
任务进度 percent 变化才推

16.3 慢客户端

Node 中 res.write() 返回 false 表示写缓冲压力较大：

javascript

const ok = res.write(payload);
if (!ok) {
  // 可以暂停读取上游，或标记慢客户端
}

策略：

可丢事件直接丢弃旧值，只保留最新快照。
不可丢事件写入消息队列，客户端断线后续读。
长时间缓冲过大就断开慢客户端。

16.4 横向扩展

单实例：

text

任务进度 -> 本机内存 -> SSE

多实例：

text

任务进度 -> Redis Streams/Kafka -> 每个 API 实例 -> 本机 SSE 连接

不要依赖本机内存保存全局连接状态。连接可以在本机，事件源要在外部可共享系统。

16.5 内部机制：正确使用 `write()` 返回值与 `drain`

Node 文档对背压的表述是明确的：res.write() 返回 false 只是建议信号——数据仍会被缓冲进内存，Node 会一直缓冲直到内存耗尽而崩溃；对 TCP socket 而言，如果对端不读数据，缓冲可能永远不 drain（浏览器 SSE 连接卡死/切后台时就会这样），这是官方文档点名的可被远程利用的风险。

正确姿势是等 drain 再写：

javascript

function writeWithBackpressure(res, payload) {
  if (res.write(payload)) return Promise.resolve();
  return new Promise((resolve, reject) => {
    const cleanup = () => {
      res.removeListener("drain", onDrain);
      res.removeListener("close", onClose);
    };
    const onDrain = () => { cleanup(); resolve(); };
    const onClose = () => { cleanup(); reject(new Error("client gone")); };
    res.once("drain", onDrain);
    res.once("close", onClose);
  });
}

对 SSE 服务端的三条务实规则：

连续 N 次 write 返回 false 即判定慢客户端：标记、降级（只推快照不推增量）或直接 res.end() 断开，让浏览器带 Last-Event-ID 重连回补。
每个连接挂一个字节水位线：监听 res.socket 的写队列或直接计数未确认输出，超过阈值（如 1MB）先停读上游。
页面隐藏时浏览器可能不再及时读流（省电策略下尤甚），后台标签页的 SSE 客户端天然偏慢——限连和降级策略要把这类误伤考虑进去，必要时在 visibilitychange 时主动断开，回前台再重连续读。

💡 高吞吐场景（日志流、AI token）宁可把“分片合并”做在上游：写侧节流聚合（每 50–100ms 刷一批）比逐个事件 write 更省 CPU、更少触发 false。

17. 测试、监控与故障排查

17.1 curl 测试

bash

curl -N http://localhost:3000/api/events

-N 表示不缓冲输出。

17.2 浏览器检查

DevTools Network：

请求状态应保持 pending。
Response 应持续追加。
Content-Type 是 text/event-stream。
是否有 CORS 错误。
是否频繁重连。

17.3 监控指标

至少监控：

当前 SSE 连接数。
每秒事件数。
连接平均持续时间。
重连次数。
客户端错误数。
服务端写失败次数。
Broker lag。
API 实例内存。
Nginx 499/502/504。

17.4 常见问题

现象	可能原因
本地实时，线上不实时	Nginx buffering、网关缓冲、CDN 缓冲
一段时间后断开	proxy_read_timeout、idle timeout、无心跳
收不到自定义事件	前端没用 `addEventListener(eventName)`
JSON 解析失败	data 不是合法 JSON，或多行格式处理错误
连接越来越多	组件卸载未 close
重复消息	重连后没有按 id 去重
不能带 Authorization header	原生 EventSource 不支持自定义 header
多实例漏推	事件只存在某个实例内存中，没有 broker

18. 学习路线与练习项目

18.1 学习路线

用 Express 写最小 SSE。
用 Vue 写 useSse。
做任务进度条。
做 AI token 流式输出。
加入 id 和 Last-Event-ID。
用 Redis Streams 支持断线续读。
配置 Nginx 关闭 buffering。
加鉴权、限流、连接数限制。
做监控和故障排查清单。

18.2 练习项目一：AI Chat SSE

技术：

Vue 3。
NestJS 或 FastAPI。
SSE。
PostgreSQL。

功能：

提交问题。
流式输出回答。
停止生成。
保存会话。
统计耗时和 token。

18.3 练习项目二：导入任务中心

技术：

Vue 3。
Spring Boot 或 NestJS。
Redis Streams。
SSE。

功能：

上传 Excel。
创建任务。
Worker 分批处理。
SSE 推进度、错误行、完成事件。
断线续读。

18.4 练习项目三：通知中心

技术：

Vue 3 + Pinia。
Node/Nest/Spring/FastAPI 任一后端。
Redis Pub/Sub 或 Redis Streams。
SSE。

功能：

实时推送通知。
未读数更新。
权限过滤。
多实例推送。
消息去重。

19. 官方参考资料

MDN Server-Sent Events：https://developer.mozilla.org/en-US/docs/Web/API/Server-sent_events
MDN Using server-sent events：https://developer.mozilla.org/en-US/docs/Web/API/Server-sent_events/Using_server-sent_events
MDN EventSource：https://developer.mozilla.org/en-US/docs/Web/API/EventSource
HTML Living Standard Server-sent events：https://html.spec.whatwg.org/multipage/server-sent-events.html
NestJS Server-Sent Events：https://docs.nestjs.com/techniques/server-sent-events
FastAPI StreamingResponse：https://fastapi.tiangolo.com/advanced/custom-response/
Spring Web MVC Async Requests：https://docs.spring.io/spring-framework/reference/web/webmvc/mvc-ann-async.html
Spring WebFlux：https://docs.spring.io/spring-framework/reference/web/webflux.html
Nginx proxy buffering：https://nginx.org/en/docs/http/ngx_http_proxy_module.html

SSE 服务端事件推送系统学习文档 ​

目录 ​

1. SSE 是什么 ​

1.1 SSE 不是什么（常见误解澄清） ​

2. SSE 和轮询、WebSocket、HTTP Stream 的区别 ​

2.1 与轮询对比 ​

2.2 与 WebSocket 对比 ​

2.3 与 HTTP Stream 对比 ​

2.4 底层差异：一条 SSE、一次 WebSocket 握手、一轮长轮询分别发生了什么 ​

3. SSE 协议格式 ​

3.1 响应头 ​

3.2 基础事件 ​

3.3 自定义事件名 ​

3.4 id 与断线续读 ​

3.5 retry ​

3.6 多行 data ​

3.7 心跳 ​

3.8 内部机制：浏览器如何逐行解析事件流 ​

3.9 内部机制：Connection: keep-alive 在 SSE 里其实是摆设 ​

4. 浏览器 EventSource ​

4.1 基础用法 ​

4.2 关闭连接 ​

4.3 Cookie 与跨域 ​

4.4 EventSource 的限制 ​

4.5 内部机制：EventSource 的状态机与自动重连规则 ​

4.6 注意事项：open 与首事件、Worker 中的 EventSource ​

5. 前端框架落地：Vue、Nuxt、React ​

5.1 Vue composable：useSse ​

5.2 Vue + Pinia：通知中心 ​

5.3 Nuxt 注意点 ​

5.4 React hook ​

6. Node.js 与 Express 实现 SSE ​

6.1 最小 SSE 接口 ​

6.2 封装 sendEvent ​

6.3 任务进度接口 ​

6.4 Node 原生 HTTP ​

7. NestJS 实现 SSE ​

7.1 @Sse 基础 ​

7.2 任务进度 ​

7.3 从 Subject 推送 ​

8. FastAPI 实现 SSE ​

8.1 StreamingResponse ​

8.2 任务进度 ​

8.3 队列驱动 ​

9. Spring Boot 实现 SSE ​

9.1 Spring MVC SseEmitter ​

9.2 Spring WebFlux ​

10. 项目落地一：AI 流式聊天 ​

10.1 需求 ​

10.2 推荐架构 ​

10.3 接口设计 ​

10.4 前端处理 ​

10.5 后端关键设计 ​

11. 项目落地二：任务进度与日志流 ​

11.1 需求 ​

11.2 架构 ​

11.3 事件设计 ​

11.4 为什么适合 SSE ​

12. 项目落地三：实时通知中心 ​

12.1 需求 ​

12.2 架构 ​

12.3 SSE 连接 ​

12.4 后端发布事件 ​

12.5 多实例问题 ​

13. 鉴权、安全与权限控制 ​

13.1 Cookie 鉴权 ​

13.2 Ticket 鉴权 ​

13.3 权限校验 ​

13.4 安全注意 ​

13.5 怎么用 fetch 读 SSE：解开自定义头的限制 ​

14. 断线重连、Last-Event-ID 与消息续读 ​

14.1 浏览器自动重连 ​

14.2 Last-Event-ID ​

14.3 续读设计 ​

14.4 前端去重 ​

14.5 内部机制：续读闭环的服务端设计 ​

15. 部署：Nginx、网关、CDN 与超时 ​

15.1 Nginx 配置 ​

15.2 网关和 CDN ​

15.3 HTTP/1.1 连接限制 ​

SSE 服务端事件推送系统学习文档

目录

1. SSE 是什么

1.1 SSE 不是什么（常见误解澄清）

2. SSE 和轮询、WebSocket、HTTP Stream 的区别

2.1 与轮询对比

2.2 与 WebSocket 对比

2.3 与 HTTP Stream 对比

2.4 底层差异：一条 SSE、一次 WebSocket 握手、一轮长轮询分别发生了什么

3. SSE 协议格式

3.1 响应头

3.2 基础事件

3.3 自定义事件名

3.4 id 与断线续读

3.5 retry

3.6 多行 data

3.7 心跳

3.8 内部机制：浏览器如何逐行解析事件流

3.9 内部机制：`Connection: keep-alive` 在 SSE 里其实是摆设

4. 浏览器 EventSource

4.1 基础用法

4.2 关闭连接

4.3 Cookie 与跨域

4.4 EventSource 的限制

4.5 内部机制：EventSource 的状态机与自动重连规则

4.6 注意事项：`open` 与首事件、Worker 中的 EventSource

5. 前端框架落地：Vue、Nuxt、React

5.1 Vue composable：useSse

5.2 Vue + Pinia：通知中心

5.3 Nuxt 注意点

5.4 React hook

6. Node.js 与 Express 实现 SSE

6.1 最小 SSE 接口

6.2 封装 sendEvent

6.3 任务进度接口

6.4 Node 原生 HTTP

7. NestJS 实现 SSE

7.1 @Sse 基础

7.2 任务进度

7.3 从 Subject 推送

8. FastAPI 实现 SSE

8.1 StreamingResponse

8.2 任务进度

8.3 队列驱动

9. Spring Boot 实现 SSE

9.1 Spring MVC SseEmitter

9.2 Spring WebFlux

10. 项目落地一：AI 流式聊天

10.1 需求

10.2 推荐架构

10.3 接口设计

10.4 前端处理

10.5 后端关键设计

11. 项目落地二：任务进度与日志流

11.1 需求

11.2 架构

11.3 事件设计

11.4 为什么适合 SSE

12. 项目落地三：实时通知中心

12.1 需求

12.2 架构

12.3 SSE 连接

12.4 后端发布事件

12.5 多实例问题

13. 鉴权、安全与权限控制

13.1 Cookie 鉴权

13.2 Ticket 鉴权

13.3 权限校验

13.4 安全注意

13.5 怎么用 fetch 读 SSE：解开自定义头的限制

14. 断线重连、Last-Event-ID 与消息续读

14.1 浏览器自动重连

14.2 Last-Event-ID

14.3 续读设计

14.4 前端去重

14.5 内部机制：续读闭环的服务端设计

15. 部署：Nginx、网关、CDN 与超时

15.1 Nginx 配置

15.2 网关和 CDN

15.3 HTTP/1.1 连接限制