流式处理与实时通信系统学习文档

核对时间：2026-07-01
适合人群：前端、Node.js、Java、Python、全栈、AI 应用、前端架构师候选人
学习目标：系统掌握轮询、长轮询、SSE、WebSocket、HTTP 流、WebRTC、消息队列、流式数据处理的选型与落地方式，并能在 Vue/Nuxt、Node/Express、NestJS、Spring Boot、FastAPI 等框架中完成真实项目实现。

为什么要学流式处理和实时通信
通信方式总览与选型表
短轮询与长轮询
SSE：服务端事件推送
WebSocket：全双工实时通信
HTTP 流与 ReadableStream
WebRTC、MQTT、gRPC Streaming 的定位
前端框架落地：Vue、Nuxt、React 思路
Node.js 与 Express 落地
NestJS 落地
Python FastAPI 落地
Java Spring Boot 落地
消息队列与流式数据处理
项目落地一：AI 流式聊天助手
项目落地二：实时通知和运营看板
项目落地三：任务进度、日志流和大文件处理
部署、网关、鉴权与横向扩展
可靠性、背压、重连与幂等
测试、监控与排查
学习路线与练习项目
官方参考资料

1. 为什么要学流式处理和实时通信

传统 HTTP 请求是“一问一答”：

text

客户端发起请求 -> 服务端处理 -> 服务端一次性返回完整响应

这适合普通 CRUD，但很多现代应用不是这样工作的：

AI 聊天：模型不是等全部生成完才返回，而是一个 token 一个 token 地吐出。
任务进度：用户上传大文件、导入 Excel、执行构建任务时，需要实时看到进度。
日志流：部署、训练、爬虫、批处理任务需要边运行边展示日志。
协同编辑：多人操作要实时同步。
IM 聊天：消息要低延迟送达。
运营看板：订单、告警、设备状态、交易数据要实时刷新。
物联网：设备数据持续上报。
流式数据处理：订单、日志、点击流进入消息系统，由消费者持续处理。

所以你要掌握两类能力：

实时通信：客户端和服务端如何实时交换数据。
流式处理：数据如何从生产者持续流向消费者，并在系统中被分发、缓冲、处理、落库、重放。

这两个能力经常一起出现：

text

浏览器
  -> SSE/WebSocket 接收实时结果
BFF / API Server
  -> 连接模型服务、任务服务、消息队列
消息队列 / 流平台
  -> Redis Streams / Kafka / RabbitMQ
Worker
  -> 消费任务、处理数据、写数据库、推送状态

2. 通信方式总览与选型表

2.1 常见方式

方式	方向	连接形态	适合场景	不适合场景
短轮询	客户端定时请求	多次 HTTP 请求	简单状态刷新、低频通知	高频实时、低延迟
长轮询	客户端请求，服务端挂起	多次长 HTTP 请求	兼容性要求高的简单实时	高并发长连接、复杂双向通信
SSE	服务端 -> 客户端	单向长连接	AI token 流、日志流、进度、通知	客户端频繁发消息给服务端
WebSocket	双向	单条长连接	IM、协同、游戏、实时看板	只需要服务端单向推送的简单场景
HTTP Chunk / Stream	服务端 -> 客户端	流式 HTTP 响应	AI 流式响应、文件流、代理上游流	需要浏览器自动重连和事件语义
WebRTC DataChannel	点对点双向	P2P，需信令	音视频、低延迟 P2P、文件点对点传输	普通后台通知
MQTT	发布/订阅	长连接	IoT、弱网设备、轻量消息	Web 后台普通业务，除非已有 broker
gRPC Streaming	双向/单向	HTTP/2	微服务内部高性能流式 RPC	直接浏览器兼容、传统前端项目
消息队列/流平台	服务间	Broker 中转	削峰、异步、重放、消费组	浏览器直接实时通信

2.2 快速选型

如果你只记一张表：

需求	优先方案
AI 聊天逐字输出	SSE 或 HTTP Stream
构建日志、任务进度	SSE
站内通知、小红点	SSE 或 WebSocket
IM 聊天、客服会话	WebSocket
协同编辑、多人画板	WebSocket + 冲突处理/CRDT
实时运营大屏	WebSocket 或 SSE
浏览器上传后看处理进度	HTTP 上传 + SSE 进度通道
微服务事件解耦	Kafka / RabbitMQ / Redis Streams
IoT 设备通信	MQTT
音视频通话	WebRTC
服务间高性能流 RPC	gRPC Streaming

2.3 不要过度设计

很多项目一上来就用 WebSocket，其实并不必要。

如果只是“用户点击生成报告，页面展示进度”，SSE 比 WebSocket 更简单：

text

POST /api/reports 创建任务
GET /api/reports/:id/events 订阅任务进度 SSE

如果只是“每 30 秒刷新一次订单状态”，短轮询就够了：

text

setInterval(() => fetch("/api/orders/status"), 30000)

架构选型要看：

是否双向。
延迟要求。
消息频率。
是否需要断线恢复。
是否需要消息持久化。
是否需要水平扩展。
网关和部署环境是否支持。
团队维护成本。

2.4 容易被忽略的选型变量：传输层协议版本

同一套 SSE/WebSocket 代码，跑在 HTTP/1.1 与 HTTP/2、HTTP/3 上，行为差别很大，选型时必须一并确认：

维度	HTTP/1.1	HTTP/2	HTTP/3 (QUIC)
一条连接上的并发请求	同域最多约 6 条连接，长连接独占槽位	单连接多路复用 stream（常见上限 100）	单连接多路复用，无队头阻塞
SSE 数量多时的表现	多标签页/多组件互相挤占，页面请求排队	每条 SSE 一个 stream，互不抢占	同 H2
中间层缓冲风险	老代理不支持流式	需网关支持 101/流式转发	部分网关/CDN 对 H3 上的流式支持有限
断线重连成本	重连=重新 TCP+TLS 握手	复用现有连接开新 stream	0-RTT 恢复、连接迁移（切网络不断线）

实操结论：

上 HTTP/2 是 SSE 免费的最大优化：解决“6 连接配额”和重连握手开销。检查方式：DevTools Network 的 Protocol 列（h2）。
WebSocket 不受影响：它握手后就脱离 HTTP 语义，H2 的 CONNECT/WebSocket-over-H2 扩展并未被广泛部署，wss:// 通常仍走 HTTP/1.1 风格的独立 TCP 连接。
弱网/频繁切网（移动 Web、小程序）优先考虑 H3 或带自动重连的 SSE，而不是自己实现复杂的连接迁移。

3. 短轮询与长轮询

3.1 短轮询

短轮询是最简单的实时方案：客户端隔一段时间请求一次。

javascript

let timer = null;

function startPolling(taskId) {
  timer = setInterval(async () => {
    const response = await fetch(`/api/tasks/${taskId}`);
    const task = await response.json();
    renderTask(task);

    if (task.status === "success" || task.status === "failed") {
      clearInterval(timer);
    }
  }, 3000);
}

优点：

实现简单。
基于普通 HTTP，几乎所有环境都支持。
方便鉴权、日志、缓存和调试。

缺点：

有延迟，取决于轮询间隔。
空请求多。
高频轮询会给服务端和数据库带来压力。

适合：

低频状态刷新。
管理后台列表刷新。
对实时性要求不高的任务状态。

3.2 长轮询

长轮询是：客户端发请求，服务端如果没有数据就先不返回，直到有数据或超时。

text

client GET /api/notifications/poll
  -> server 等待消息
  -> 有消息：返回消息
  -> 没消息：超时返回空
client 收到响应后立刻再次请求

Node 示例：

javascript

app.get("/api/notifications/poll", async (req, res) => {
  const userId = req.user.id;
  const message = await waitMessage(userId, 25_000);

  if (!message) {
    res.json({ messages: [] });
    return;
  }

  res.json({ messages: [message] });
});

长轮询适合兼容性要求高、不能使用 SSE/WebSocket 的场景。但现代 Web 项目中，如果需要服务端单向推送，通常优先 SSE；如果需要双向实时，优先 WebSocket。

3.3 内部机制：长轮询占用的到底是什么资源

长轮询看起来“还是 HTTP”，但挂起的每个请求都真实占用：

text

一个挂起的长轮询 = 1 个网关连接槽 + 1 个服务端 socket/任务上下文
                  + （常见反模式）1 个被阻塞的线程或 1 个挂起协程

理解这几点才能解释它的局限：

维度	机制
连接配额	HTTP/1.1 下浏览器对同一域名并发连接数有限（经典值 6）。长轮询在挂起期间独占一个槽位，页面里其他 XHR/fetch 会被排队——这是“开了轮询页面其他请求变慢”的根因
服务端模型	同步线程模型（传统 Servlet/PHP-FPM）下挂起=占线程，几百个并发挂起就能耗尽 worker；Node/异步模型只占句柄不占线程，代价小得多
超时博弈	服务端挂起时长必须小于链路上最短的 idle timeout（网关常见 60s），否则代理先掐、客户端收到假“网络错误”。约定 25–30s 返回空响应再重发是经验值
消息时序	响应-再请求的间隙里到达的消息靠服务端缓冲队列兜住；一次只推一条会放大往返次数，应批量返回

⚠️ 长轮询与浏览器缓存/预取冲突：轮询 URL 若带随机参数可避开 HTTP 缓存但会污染 CDN 命中率；若被中间代理缓存，消息直接不可达。生产上轮询接口一律 Cache-Control: no-store。

💡 为什么现代项目基本淘汰长轮询：SSE 把“挂起-重连-重发请求”的循环交给浏览器原生实现（自动重连 + Last-Event-ID），且一条连接持续复用，省掉了每轮 HTTP 事务头部与重新鉴权；只有在不支持 EventSource 或需要模拟上行高频发送（WebSocket 被防火墙封锁、只剩 HTTP 通道，如早期 Socket.js/SockJS 的 xhr_polling 传输）时才值得。

4. SSE：服务端事件推送

SSE 是 Server-Sent Events，浏览器通过 EventSource 建立一个 HTTP 长连接，服务端持续推送文本事件。

4.1 SSE 数据格式

响应头：

http

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

事件格式：

text

id: 1001
event: progress
data: {"percent":50,"message":"processing"}

注意最后有一个空行，用于表示事件结束。

常见字段：

字段	作用
`data`	事件内容，可以多行
`event`	自定义事件名
`id`	事件 ID，断线重连时可配合 `Last-Event-ID`
`retry`	建议客户端重连间隔

4.2 浏览器 EventSource

javascript

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

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

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

source.onerror = () => {
  console.log("SSE disconnected, browser may retry automatically");
};

优点：

浏览器原生支持。
自动重连。
基于 HTTP，服务端实现简单。
适合文本流、进度、日志、AI token。

缺点：

浏览器原生 EventSource 主要是 GET，不适合带复杂请求体。
单向：服务端推客户端；客户端要发消息仍需普通 HTTP 请求。
老旧代理或网关可能缓冲响应，需要专门配置。
HTTP/1.1 下浏览器对同域连接数有限制。

4.3 SSE 适合 AI 流式输出

AI 聊天典型流程：

text

POST /api/chat 创建会话或提交消息
GET /api/chat/:messageId/stream 订阅模型输出

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

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

event: done
data: {"messageId":"m_123"}

为什么不用 WebSocket？

AI 输出通常是服务端单向推送。
用户输入可以通过普通 POST 提交。
SSE 自动重连和 HTTP 语义更简单。
网关、日志、鉴权链路更接近普通 HTTP。

如果聊天还需要“多人在线、正在输入、客服坐席状态、双向消息”，可以升级为 WebSocket。

4.4 注意事项：EventSource 的三条容易踩空的规范行为

以下行为来自 HTML 标准对 EventSource 的规定（详见 SSE 专章文档 4.5 节的状态机分析），在选型评审时经常被误以为“浏览器 bug”：

非 200 或 Content-Type 不是 text/event-stream 时，连接直接失败且不自动重连。网关把 SSE 改写成 200+text/plain（比如某些 WAF 的“错误页替换”）会让 EventSource 静默死掉。
204 No Content 是唯一能让服务端“劝退”浏览器停止重连的响应——做“会话结束请别再连了”用 204，而不是简单 res.end()（后者浏览器会按 retry 间隔一直重连）。
流只按 UTF-8 解码，没有第二种编码；重定向（301/307）可以正常跟随，跨源需要标准 CORS 头。

另外，HTTP/1.1 的“同域 6 连接”限制是按浏览器全局计数的（所有标签页共享），多标签页 SSE 场景要么合并为单连接（BroadcastChannel 分发），要么上 HTTP/2，参见 2.4 节。

5. WebSocket：全双工实时通信

WebSocket 建立后，客户端和服务端可以在同一条连接上双向发送消息。

5.1 浏览器 WebSocket

javascript

const socket = new WebSocket("wss://example.com/ws");

socket.onopen = () => {
  socket.send(JSON.stringify({
    type: "join_room",
    roomId: "room_1",
  }));
};

socket.onmessage = (event) => {
  const message = JSON.parse(event.data);
  handleMessage(message);
};

socket.onclose = () => {
  console.log("socket closed");
};

socket.onerror = (error) => {
  console.error("socket error", error);
};

5.2 消息协议设计

不要直接发送随意字符串。项目中应设计统一消息协议：

typescript

type RealtimeMessage =
  | {
      type: "chat.message";
      requestId: string;
      roomId: string;
      payload: {
        messageId: string;
        content: string;
        senderId: string;
        createdAt: string;
      };
    }
  | {
      type: "notification.created";
      requestId: string;
      payload: {
        id: string;
        title: string;
      };
    }
  | {
      type: "ping";
      requestId: string;
      payload: Record<string, never>;
    };

协议字段建议：

字段	作用
`type`	消息类型
`requestId`	排查、幂等、链路追踪
`roomId` / `channel`	频道或房间
`payload`	业务数据
`createdAt`	服务端时间
`version`	协议版本

5.3 WebSocket 适合什么

适合：

IM 聊天。
多人协作。
实时游戏。
客服系统。
实时看板高频更新。
客户端也要频繁给服务端发消息。

不适合：

只是服务端单向推送几个进度事件。
服务端和网关没有长连接维护能力。
业务团队没有设计重连、心跳、鉴权和扩展的经验。

5.4 心跳与重连

浏览器侧需要重连：

javascript

function createSocket(url) {
  let socket = null;
  let retry = 0;
  let heartbeatTimer = null;

  function connect() {
    socket = new WebSocket(url);

    socket.onopen = () => {
      retry = 0;
      heartbeatTimer = setInterval(() => {
        socket.send(JSON.stringify({ type: "ping" }));
      }, 30000);
    };

    socket.onmessage = (event) => {
      handleMessage(JSON.parse(event.data));
    };

    socket.onclose = () => {
      clearInterval(heartbeatTimer);
      const delay = Math.min(1000 * 2 ** retry, 30000);
      retry += 1;
      setTimeout(connect, delay);
    };
  }

  connect();

  return {
    send(message) {
      if (socket?.readyState === WebSocket.OPEN) {
        socket.send(JSON.stringify(message));
      }
    },
    close() {
      clearInterval(heartbeatTimer);
      socket?.close();
    },
  };
}

生产项目还要处理：

token 过期。
网络恢复。
页面隐藏和恢复。
多标签页重复连接。
消息去重。
未发送消息缓存。

5.5 内部机制：握手与帧格式

WebSocket 的“看起来像 HTTP、用起来不是 HTTP”发生在一次握手之后（RFC 6455）：

text

客户端                                      服务端
GET /ws HTTP/1.1
Upgrade: websocket                ──────>  校验 Upgrade/Connection 头
Connection: Upgrade                        生成 accept key =
Sec-WebSocket-Key: <随机16字节的Base64>       base64(SHA1(key + GUID))
Sec-WebSocket-Version: 13                  （GUID 固定为
                                        258EAFA5-E914-47DA-95CA-C5AB0DC85B11）
                             <──────────  HTTP/1.1 101 Switching Protocols
                                        Upgrade: websocket
                                        Sec-WebSocket-Accept: <上面结果>
========== 此后这条 TCP 上不再有 HTTP，只有 WebSocket 帧 ==========

握手要点：

Sec-WebSocket-Key/Accept 不是加密，只是“证明对方真的理解 WebSocket”的握手探测，防止普通 HTTP 服务器误答 101。
网关必须透传 Upgrade、Connection、Sec-WebSocket-* 头并支持 101；返回 200/400/426 基本都是这层没配好。
Sec-WebSocket-Protocol 可协商子协议，常被“借用”传鉴权 token（如 ['bearer', token]），若 101 响应缺少该头表示服务端未接受任何候选子协议，握手按失败处理。

帧（frame）层面决定性能与排错：

字段/机制	说明
opcode	`0x1` 文本、`0x2` 二进制、`0x8` close、`0x9` ping、`0xA` pong、`0x0` 续帧
FIN + 续帧	一条消息可拆多帧发送（分片），接收端按 FIN 组包；大消息分片会增加服务端内存驻留
masking	客户端→服务端帧必须掩码（防代理缓存投毒等历史攻击）；服务端→客户端不掩码。服务端 CPU 解密开销在客户端侧
控制帧	ping/pong/close 不可分片、payload ≤ 125 字节；收到 Ping 必须尽快回同 payload 的 Pong（这也是 5.4 心跳的底层依据，浏览器 `WebSocket` API 不暴露 ping/pong，需应用层自定义消息）
帧头开销	短消息仅 2 字节（服务端发出）/6 字节（客户端发出，含 mask key）；payload >125 字节时长度扩展字段再加 2 或 8 字节
close code	`1000` 正常、`1006` 异常断开（无 close 帧，JS 里拿不到原因）、`1009` 消息过大、`1013` 忙/稍后再试；重连逻辑应按 code 区分“该不该重试”

⚠️ 1006 是排查中最常见也最无信息的码：TLS 失败、网关掐断、握手被拒都会表现为它，真实原因要看服务端日志与 DevTools 的 Network → WS → 握手请求响应头。

💡 WebSocket 没有应用层 ACK。TCP 只保证送达内核，不保证业务处理成功；可靠消息必须靠自己的 requestId/序号 + 应用层确认 + 重发，或干脆把持久化交给消息队列（第 13 章）。

6. HTTP 流与 ReadableStream

SSE 是有事件格式的文本流；HTTP Stream 更通用，适合代理上游模型输出、文件流、二进制流或自定义协议。

6.1 fetch 读取流

javascript

async function streamText() {
  const response = await fetch("/api/chat/stream", {
    method: "POST",
    headers: {
      "Content-Type": "application/json",
    },
    body: JSON.stringify({ message: "解释一下 Vue 响应式" }),
  });

  if (!response.body) {
    throw new Error("ReadableStream is not supported");
  }

  const reader = response.body.getReader();
  const decoder = new TextDecoder();
  let result = "";

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

    result += decoder.decode(value, { stream: true });
    renderPartialText(result);
  }
}

6.2 SSE vs HTTP Stream

对比	SSE	HTTP Stream
浏览器 API	`EventSource`	`fetch().body.getReader()`
请求方法	原生主要 GET	GET/POST 都可
消息格式	`event/data/id` 文本协议	自定义
自动重连	浏览器支持	需要自己实现
适合	事件流、进度、日志	模型代理、二进制、自定义流

AI 场景中，如果需要 POST 请求体直接发起并读取流，HTTP Stream 更方便；如果先创建任务再订阅事件，SSE 更清晰。

6.3 内部机制：ReadableStream 的队列、水位线与背压

response.body 是一个符合 WHATWG Streams 标准的 ReadableStream。理解它的三个内部构件，才能写出不堆内存的流式消费代码：

text

underlying source（网络字节）
   │ controller.enqueue(chunk)
   ▼
┌──────────────────────────┐
│ internal queue            │   highWaterMark（水位线）
│ [chunk][chunk][chunk]...  │   desiredSize = HWM - Σqueue.size
└──────────────────────────┘
   │ reader.read()
   ▼
consumer（你的 while 循环 / pipeTo 目标）

内部队列 + 排队策略：chunk 入队但未被读走时驻留在内部队列。构造流时可传 queuingStrategy（CountQueuingStrategy 按块数、ByteLengthQueuingStrategy 按字节数），核心参数是 highWaterMark。
desiredSize 信号：controller.desiredSize = highWaterMark - 队列内 chunk 总量。它降到 ≤ 0 时，浏览器停止调用 source 的 pull()——这就是读侧背压：消费者慢，取数就停。对 fetch 响应流，这最终表现为 TCP 接收窗口不再扩张，把压力顶回服务端。
pull 模型：ReadableStream 的 source 有 start()（一次性）和 pull(controller)（每当队列需要更多数据时被调用，且前一次 pull 的 Promise resolve 之前不会并发再调）。自己桥接 WebSocket/事件源到流时，把“等下一事件”放进 pull 返回的 Promise，而不是无脑 enqueue——后者会把事件全堆进内存。

javascript

// 用 pull 正确桥接事件源：一次 pull 只等一个事件，天然背压
function readableFromEvent(emitter, event) {
  const buffered = [];   // 无人等待时暂存的事件（生产要设上限，超限丢最旧）
  let waiting = null;    // 当前挂起的 pull 的投递函数

  function onEvent(payload) {
    if (waiting) { const deliver = waiting; waiting = null; deliver(payload); }
    else buffered.push(payload);
  }
  emitter.on(event, onEvent);

  return new ReadableStream({
    pull(controller) {
      if (buffered.length) { controller.enqueue(buffered.shift()); return; }
      // 队列空：返回 Promise，等下一个事件到达再 enqueue 并放行
      return new Promise((resolve) => {
        waiting = (payload) => { controller.enqueue(payload); resolve(); };
      });
    },
    cancel() { emitter.off(event, onEvent); },
  });
}

配套要点：

reader.read() 每次取一个 chunk；read(n)（getReader({ mode: "byob" })，BYOB 读）可要求“凑满 n 字节再给”，适合定长协议解析，直接写入自备 buffer、减少拷贝。
pipeTo()/pipeThrough() 自带背压传播：下游 Writable 的队列满，上游自动暂停。手动 while 循环里则靠“await read() 之前不取下一块”实现同样效果——快慢匹配，不排队。
流只能被一个 reader 锁定消费；要同时“渲染 + 缓存”，用 tee() 分成两路。⚠️ tee() 两路消费速度不一致时，快路要等慢路，差值全部积压在内存——这是“流式页面越开越耗内存”的经典原因。
for await...of stream 已标准化（Chrome 124+/Firefox 110+）：break 退出默认会 cancel 整个流；想中途暂停稍后续读要 stream.values({ preventCancel: true })。

6.4 注意事项：解码与切帧必须在消费侧做

response.body 给你的是任意边界的字节 chunk——一个 chunk 可能包含半个多字节字符、半行 SSE、半条 JSON。两件事必须自己做：

UTF-8 跨块解码：decoder.decode(value, { stream: true })（或直接 pipeThrough(new TextDecoderStream())），漏掉 stream: true 会把块边界上的多字节字符解成 U+FFFD 乱码。
应用层切帧：网络 chunk 边界 ≠ 消息边界。NDJSON 按 \n 切、SSE 按空行切、长度前缀协议按头切；残帧留在自己的字符串 buffer 里等下一 chunk。AI 流式“偶发 JSON.parse 失败”几乎都是直接 JSON.parse(chunk) 导致。

javascript

// NDJSON 残帧缓冲的最小正确实现
let buf = "";
for await (const chunk of response.body.pipeThrough(new TextDecoderStream())) {
  buf += chunk;
  let i;
  while ((i = buf.indexOf("\n")) >= 0) {
    const line = buf.slice(0, i).trim();
    buf = buf.slice(i + 1);
    if (line) handleMessage(JSON.parse(line));
  }
}

7. WebRTC、MQTT、gRPC Streaming 的定位

7.1 WebRTC

WebRTC 主要用于浏览器之间的低延迟音视频、数据通道。它通常需要信令服务器交换 SDP 和 ICE candidate。

典型结构：

text

Browser A
  -> WebSocket 信令服务器
Browser B
  -> WebSocket 信令服务器

信令完成后：
Browser A <-> Browser B 直接 P2P 通信

适合：

音视频通话。
屏幕共享。
点对点文件传输。
低延迟互动。

不适合普通后台通知或 AI token 输出。

7.2 MQTT

MQTT 是轻量发布订阅协议，常用于 IoT。

text

device/001/status
device/001/telemetry
device/001/command

适合：

设备弱网。
低功耗。
海量设备上报。
发布订阅模型。

Web 前端可以通过 MQTT over WebSocket 接入，但普通 Web 项目没必要为了实时通知引入 MQTT，除非后端和设备系统已经以 MQTT 为核心。

7.3 gRPC Streaming

gRPC Streaming 适合服务间通信：

Server streaming：服务端持续返回。
Client streaming：客户端持续发送。
Bidirectional streaming：双向流。

适合：

微服务内部。
高性能 RPC。
统一 Protobuf 协议。
后端系统之间持续同步数据。

浏览器直接使用会有额外兼容成本，前端项目通常通过 BFF 转成 SSE/WebSocket/HTTP Stream。

8. 前端框架落地：Vue、Nuxt、React 思路

8.1 Vue：封装 useSse

typescript

import { onBeforeUnmount, ref } from "vue";

export function useSse<T>(url: string, eventName = "message") {
  const data = ref<T | null>(null);
  const error = ref<Event | null>(null);
  const connected = ref(false);

  let source: EventSource | null = null;

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

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

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

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

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

  onBeforeUnmount(close);

  connect();

  return {
    data,
    error,
    connected,
    close,
  };
}

任务进度组件：

vue

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

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

const { data, connected } = useSse<{
  percent: number;
  message: string;
}>(`/api/tasks/${props.taskId}/events`, "progress");

const percent = computed(() => data.value?.percent ?? 0);
</script>

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

8.2 Vue：封装 useWebSocket

typescript

import { onBeforeUnmount, ref } from "vue";

export function useWebSocket(url: string) {
  const connected = ref(false);
  const messages = ref<any[]>([]);
  let socket: WebSocket | null = null;
  let retry = 0;
  let closedByUser = false;

  function connect() {
    socket = new WebSocket(url);

    socket.onopen = () => {
      connected.value = true;
      retry = 0;
    };

    socket.onmessage = (event) => {
      messages.value.push(JSON.parse(event.data));
    };

    socket.onclose = () => {
      connected.value = false;
      if (closedByUser) return;
      const delay = Math.min(1000 * 2 ** retry, 30000);
      retry += 1;
      setTimeout(connect, delay);
    };
  }

  function send(type: string, payload: unknown) {
    if (socket?.readyState !== WebSocket.OPEN) return false;
    socket.send(JSON.stringify({
      type,
      requestId: crypto.randomUUID(),
      payload,
    }));
    return true;
  }

  function close() {
    closedByUser = true;
    socket?.close();
  }

  connect();
  onBeforeUnmount(close);

  return {
    connected,
    messages,
    send,
    close,
  };
}

8.3 Pinia 管理实时状态

typescript

import { defineStore } from "pinia";

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

WebSocket 消息进入 store：

typescript

const notificationStore = useNotificationStore();

socket.onmessage = (event) => {
  const message = JSON.parse(event.data);
  if (message.type === "notification.created") {
    notificationStore.addNotification(message.payload);
  }
};

8.4 Nuxt 落地注意点

Nuxt 项目要区分：

客户端连接：在 onMounted 后创建 EventSource / WebSocket。
服务端路由：可作为 BFF 代理上游流。
SSR 阶段不能直接访问 window、WebSocket、EventSource。

客户端插件：

typescript

export default defineNuxtPlugin(() => {
  if (import.meta.server) return;

  return {
    provide: {
      createRealtimeSocket(url: string) {
        return new WebSocket(url);
      },
    },
  };
});

8.5 React 思路

React 中把 Vue composable 换成 hook：

tsx

import { useEffect, useRef, useState } from "react";

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

  useEffect(() => {
    const source = new EventSource(url);
    sourceRef.current = source;

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

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

  return data;
}

核心思想一样：连接生命周期必须绑定组件生命周期，卸载时清理。

9. Node.js 与 Express 落地

9.1 Express SSE

javascript

import express from "express";

const app = express();

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.flushHeaders?.();

  const send = (event, data, id) => {
    if (id) res.write(`id: ${id}\n`);
    res.write(`event: ${event}\n`);
    res.write(`data: ${JSON.stringify(data)}\n\n`);
  };

  send("connected", { taskId });

  const timer = setInterval(async () => {
    const task = await getTask(taskId);
    send("progress", {
      percent: task.percent,
      message: task.message,
    });

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

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

落地要点：

必须设置 text/event-stream。
每个事件以空行结束。
客户端断开时清理 timer、订阅、队列消费者。
代理层关闭缓冲。
不要在 SSE 连接里直接做重 CPU 任务。

9.2 Express + HTTP Stream

javascript

app.post("/api/chat/stream", express.json(), async (req, res) => {
  res.setHeader("Content-Type", "text/plain; charset=utf-8");
  res.setHeader("Cache-Control", "no-cache, no-transform");

  for await (const chunk of callModelStream(req.body.message)) {
    res.write(chunk.text);
  }

  res.end();
});

如果上游本身就是 Node stream，使用 pipeline 更安全：

javascript

import { pipeline } from "node:stream/promises";

app.get("/api/files/:id/download", async (req, res) => {
  const stream = await createFileReadStream(req.params.id);
  res.setHeader("Content-Type", "application/octet-stream");
  await pipeline(stream, res);
});

9.3 ws WebSocket Server

javascript

import http from "node:http";
import { WebSocketServer } from "ws";

const server = http.createServer(app);
const wss = new WebSocketServer({ server, path: "/ws" });

const rooms = new Map();

wss.on("connection", (socket, request) => {
  socket.on("message", (raw) => {
    const message = JSON.parse(raw.toString());

    if (message.type === "join_room") {
      const room = rooms.get(message.roomId) ?? new Set();
      room.add(socket);
      rooms.set(message.roomId, room);
      return;
    }

    if (message.type === "chat.message") {
      const room = rooms.get(message.roomId);
      if (!room) return;

      for (const client of room) {
        if (client.readyState === client.OPEN) {
          client.send(JSON.stringify(message));
        }
      }
    }
  });

  socket.on("close", () => {
    for (const room of rooms.values()) {
      room.delete(socket);
    }
  });
});

server.listen(3000);

生产项目不要只用内存 rooms，多实例部署时要接 Redis Pub/Sub 或 Socket.IO Redis Adapter，否则 A 实例上的用户收不到 B 实例发布的消息。

10. NestJS 落地

10.1 NestJS SSE

NestJS 提供 @Sse()，通常返回 RxJS Observable。

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,
      })),
    );
  }
}

适合：

任务进度。
AI token 转发。
通知推送。
审批流状态。

如果数据来自消息队列，可以把 Redis/Kafka 消费结果转成 Observable。

10.2 NestJS WebSocket Gateway

typescript

import {
  ConnectedSocket,
  MessageBody,
  SubscribeMessage,
  WebSocketGateway,
  WebSocketServer,
} from "@nestjs/websockets";
import { Server, Socket } from "socket.io";

@WebSocketGateway({
  cors: {
    origin: ["https://example.com"],
    credentials: true,
  },
})
export class ChatGateway {
  @WebSocketServer()
  server: Server;

  @SubscribeMessage("room.join")
  async joinRoom(
    @ConnectedSocket() client: Socket,
    @MessageBody() body: { roomId: string },
  ) {
    await client.join(body.roomId);
    return { ok: true };
  }

  @SubscribeMessage("chat.message")
  async sendMessage(
    @ConnectedSocket() client: Socket,
    @MessageBody() body: { roomId: string; content: string },
  ) {
    const message = await saveMessage({
      roomId: body.roomId,
      content: body.content,
      senderId: client.data.userId,
    });

    this.server.to(body.roomId).emit("chat.message", message);
  }
}

10.3 NestJS 鉴权

WebSocket 不能完全照搬 HTTP 请求拦截器。通常在握手阶段校验 token：

typescript

async handleConnection(client: Socket) {
  const token = client.handshake.auth?.token;
  const user = await this.authService.verifyToken(token);

  if (!user) {
    client.disconnect(true);
    return;
  }

  client.data.userId = user.id;
}

落地要点：

握手鉴权。
room 权限校验。
消息 DTO 校验。
限流。
Redis adapter 支持多实例广播。
断开时释放资源。

11. Python FastAPI 落地

11.1 StreamingResponse

python

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

app = FastAPI()

async def token_generator(message: str):
    for token in ["你好", "，", "这是", "流式", "响应"]:
        await asyncio.sleep(0.2)
        yield token

@app.post("/api/chat/stream")
async def chat_stream(payload: dict):
    return StreamingResponse(
        token_generator(payload["message"]),
        media_type="text/plain; charset=utf-8",
    )

SSE 格式：

python

import json

async def sse_generator(task_id: str):
    for i in range(101):
        await asyncio.sleep(0.1)
        data = json.dumps({"percent": i}, ensure_ascii=False)
        yield f"event: progress\ndata: {data}\n\n"

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

11.2 FastAPI WebSocket

python

from fastapi import FastAPI, WebSocket, WebSocketDisconnect

app = FastAPI()

class ConnectionManager:
    def __init__(self):
        self.rooms: dict[str, set[WebSocket]] = {}

    async def connect(self, room_id: str, websocket: WebSocket):
        await websocket.accept()
        self.rooms.setdefault(room_id, set()).add(websocket)

    def disconnect(self, room_id: str, websocket: WebSocket):
        self.rooms.get(room_id, set()).discard(websocket)

    async def broadcast(self, room_id: str, message: dict):
        for websocket in list(self.rooms.get(room_id, set())):
            await websocket.send_json(message)

manager = ConnectionManager()

@app.websocket("/ws/rooms/{room_id}")
async def room_socket(websocket: WebSocket, room_id: str):
    await manager.connect(room_id, websocket)
    try:
      while True:
          message = await websocket.receive_json()
          await manager.broadcast(room_id, message)
    except WebSocketDisconnect:
      manager.disconnect(room_id, websocket)

生产注意：

单进程内存 manager 不能跨实例。
多 worker 部署要接 Redis Pub/Sub、Kafka 或专门实时网关。
async endpoint 里不要执行长时间阻塞同步函数。
对每个连接限制消息大小和频率。

11.3 FastAPI + 队列推送任务进度

python

import asyncio

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

async def run_import_task(task_id: str):
    queue = task_queues[task_id]
    for percent in range(0, 101, 10):
        await asyncio.sleep(0.5)
        await queue.put({"event": "progress", "percent": percent})
    await queue.put({"event": "done"})

async def task_event_stream(task_id: str):
    queue = task_queues.setdefault(task_id, asyncio.Queue())
    while True:
        item = await queue.get()
        yield f"event: {item['event']}\ndata: {json.dumps(item)}\n\n"
        if item["event"] == "done":
            break

这个模型适合单实例学习。生产要把队列换成 Redis Streams、Kafka、数据库状态表或任务系统。

12. Java Spring Boot 落地

12.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.watchProgress(taskId, progress -> {
            try {
                emitter.send(SseEmitter.event()
                    .name("progress")
                    .data(progress));

                if (progress.finished()) {
                    emitter.send(SseEmitter.event()
                        .name("done")
                        .data(progress));
                    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 项目。

12.2 Spring WebFlux SSE

WebFlux 天然适合响应式流：

java

@RestController
@RequestMapping("/api/tasks")
public class TaskFluxController {
    private final 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/Reactor。
上游数据源也是响应式。

不要在 WebFlux 链路里直接调用阻塞 JDBC 或阻塞 SDK。必须调用时，把阻塞任务放到合适调度器或隔离服务。

12.3 Spring WebSocket / STOMP

后台管理系统或 IM 可以使用 WebSocket + STOMP：

java

@Configuration
@EnableWebSocketMessageBroker
public class WebSocketConfig implements WebSocketMessageBrokerConfigurer {
    @Override
    public void configureMessageBroker(MessageBrokerRegistry registry) {
        registry.enableSimpleBroker("/topic", "/queue");
        registry.setApplicationDestinationPrefixes("/app");
    }

    @Override
    public void registerStompEndpoints(StompEndpointRegistry registry) {
        registry.addEndpoint("/ws")
            .setAllowedOriginPatterns("https://example.com")
            .withSockJS();
    }
}

Controller：

java

@Controller
public class ChatController {
    @MessageMapping("/chat.send")
    @SendTo("/topic/rooms/{roomId}")
    public ChatMessage send(ChatMessage message) {
        return message;
    }
}

生产项目需要：

鉴权拦截。
用户订阅权限。
broker relay 或外部消息系统。
心跳。
消息持久化。
离线消息策略。

13. 消息队列与流式数据处理

实时通信解决“怎么推给客户端”，消息队列解决“服务之间怎么可靠传递和处理”。

13.1 Pub/Sub 与 Stream 的区别

能力	Redis Pub/Sub	Redis Streams	Kafka
消息持久化	否	是	是
离线消费	否	是	是
消费组	否	是	是
重放	否	有限	强
延迟	低	低	低到中
适合	在线广播	任务进度、轻量事件流	大规模日志、订单、点击流

13.2 Redis Pub/Sub

适合 WebSocket 多实例广播：

text

API 实例 A 收到消息
  -> publish room:1 message
API 实例 B/C 订阅 room:1
  -> 推给自己实例上的 WebSocket 客户端

问题：消费者离线时消息丢失，所以不适合重要业务消息持久化。

13.3 Redis Streams

适合任务进度和轻量流式处理：

text

Worker XADD task:123:events * event progress percent 50
API SSE XREAD BLOCK 30000 STREAMS task:123:events lastId
Browser EventSource 接收 progress

优点：

可阻塞读取。
有消息 ID。
可断点续读。
可消费组处理。

13.4 Kafka

Kafka 适合大规模事件流：

text

orders.created
logs.ingested
user.behavior.clicked
metrics.raw

典型处理链路：

text

API Server
  -> 写 Kafka topic
Stream Worker
  -> 消费 topic
  -> 清洗/聚合/告警
  -> 写数据库/缓存
Realtime Gateway
  -> 推送关键事件到 WebSocket/SSE

适合：

高吞吐。
需要重放。
多消费者独立处理。
日志、点击流、交易流。

13.5 RabbitMQ

RabbitMQ 更适合传统任务队列和复杂路由：

下单后发邮件。
生成报表。
导入文件。
延迟任务。
重试和死信队列。

实时推送时通常是：

text

业务服务 -> RabbitMQ -> Worker -> WebSocket/SSE Gateway -> 浏览器

14. 项目落地一：AI 流式聊天助手

14.1 需求

做一个企业 AI 助手：

用户输入问题。
前端逐字显示模型输出。
支持停止生成。
支持失败重试。
保存完整消息。
管理员可查看调用耗时、token、错误。

14.2 推荐架构

text

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

BFF / API Server
  -> 校验登录和会话权限
  -> 调用模型服务流式 API
  -> 边接收边写入 SSE
  -> 最终保存完整 assistant message

Database
  -> conversations
  -> messages
  -> model_call_logs

14.3 为什么选 SSE

输出是服务端单向推送。
用户输入用普通 POST 更好做鉴权和审计。
浏览器自动重连。
与 HTTP 网关和 BFF 更容易集成。

14.4 接口设计

创建消息：

http

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

{
  "content": "帮我解释一下 WebSocket 和 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}}

14.5 前端落地

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 = createEmptyAssistantMessage(messageId);

  const source = new EventSource(streamUrl);

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

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

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

14.6 后端落地重点

必须处理：

用户是否有权限访问 conversation。
上游模型超时。
用户点击停止生成。
浏览器断开后是否继续生成。
生成中内容如何暂存。
最终消息如何保存。
敏感词和内容安全。
费用统计和限流。

停止生成：

text

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

服务端用 AbortController 或框架对应取消机制中止上游调用。

14.7 数据库表建议

text

conversations
  id
  user_id
  title
  created_at
  updated_at

messages
  id
  conversation_id
  role
  content
  status
  created_at

model_call_logs
  id
  message_id
  model
  input_tokens
  output_tokens
  latency_ms
  error_code

15. 项目落地二：实时通知和运营看板

15.1 需求

后台管理系统需要：

新订单实时提醒。
风险告警弹窗。
顶部通知小红点。
运营大屏实时数字更新。
多个服务都能发布通知。

15.2 推荐架构

text

Order Service / Risk Service / Ops Service
  -> Redis Pub/Sub 或 Kafka
Realtime Gateway
  -> WebSocket/SSE
Vue Admin
  -> Pinia notification store

如果只需要服务端推浏览器：

通知小红点：SSE 足够。
高频大屏：WebSocket 更灵活。

如果客户端也要发订阅条件、筛选条件、房间消息：

使用 WebSocket。

15.3 WebSocket 房间设计

text

tenant:{tenantId}
tenant:{tenantId}:user:{userId}
tenant:{tenantId}:dashboard:{dashboardId}

消息：

json

{
  "type": "order.created",
  "requestId": "req_001",
  "channel": "tenant:1:dashboard:main",
  "payload": {
    "orderId": "o_1001",
    "amount": 199
  },
  "createdAt": "2026-07-01T10:00:00.000Z"
}

15.4 前端 Pinia 接入

typescript

const socket = createSocket("/ws");
const notificationStore = useNotificationStore();
const dashboardStore = useDashboardStore();

socket.on("message", (message) => {
  switch (message.type) {
    case "notification.created":
      notificationStore.addNotification(message.payload);
      break;
    case "dashboard.metric.updated":
      dashboardStore.updateMetric(message.payload);
      break;
  }
});

15.5 后端多实例广播

不要这样：

text

Instance A 内存 rooms
Instance B 内存 rooms

因为用户连接到 B，消息发布到 A 时，B 不知道。

正确方式：

text

Instance A publish Redis channel
Instance B subscribe Redis channel
Instance B 推给本机 WebSocket clients

或者使用 Socket.IO Redis Adapter、Kafka consumer group、专门实时网关。

16. 项目落地三：任务进度、日志流和大文件处理

16.1 需求

用户上传 Excel，后台导入数据：

文件上传可能耗时。
解析、校验、入库需要进度。
错误行要实时显示。
导入完成后下载错误报告。

16.2 推荐架构

text

Vue
  -> POST /api/imports 上传文件，返回 taskId
  -> GET /api/imports/:taskId/events SSE 监听进度

API Server
  -> 保存文件
  -> 创建 import_task
  -> 投递任务到队列

Worker
  -> 读取文件
  -> 分批校验
  -> 分批入库
  -> 写进度到 Redis Streams / DB

SSE Endpoint
  -> 读取 Redis Streams / DB
  -> 推给浏览器

16.3 事件设计

json

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

错误行：

json

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

完成：

json

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

16.4 为什么不用 WebSocket

任务进度和日志通常是服务端单向推送，SSE 更简单：

上传用 HTTP。
进度用 SSE。
错误报告下载用 HTTP。

WebSocket 不是不能做，而是会增加连接管理、房间、心跳、重连和扩展复杂度。

17. 部署、网关、鉴权与横向扩展

17.1 Nginx SSE 配置

SSE 需要关闭代理缓冲，否则服务端写了数据，浏览器可能迟迟收不到。

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;
}

服务端也可加：

http

X-Accel-Buffering: no

17.2 Nginx WebSocket 配置

nginx

location /ws/ {
  proxy_pass http://realtime_upstream;
  proxy_http_version 1.1;
  proxy_set_header Upgrade $http_upgrade;
  proxy_set_header Connection "upgrade";
  proxy_set_header Host $host;
  proxy_read_timeout 3600s;
}

17.3 鉴权

SSE：

Cookie session：浏览器 EventSource 可自动带 cookie。
token：原生 EventSource 不能直接设置自定义 header，可通过 cookie、短期 ticket 或 query token。

WebSocket：

握手 query。
Sec-WebSocket-Protocol。
Socket.IO auth。
首条消息鉴权。

17.4 横向扩展

长连接服务多实例部署时，需要解决：

连接在哪个实例上。
消息从哪个实例发布。
如何广播到其他实例。
实例下线如何迁移连接。

方案：

问题	方案
同一用户连接到不同实例	无状态鉴权 + 任意实例可连接
跨实例广播	Redis Pub/Sub、Kafka、Socket.IO Adapter
房间状态	外部存储或每实例本地 + broker 广播
连接数过多	独立 realtime gateway
会话粘性	可用 sticky session，但不应成为唯一方案

18. 可靠性、背压、重连与幂等

18.1 断线恢复

SSE 支持 id 和 Last-Event-ID：

text

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

浏览器重连时会带上 Last-Event-ID。服务端可以从 Redis Streams 或数据库读取缺失事件。

WebSocket 需要自己实现：

json

{
  "type": "resume",
  "payload": {
    "lastMessageId": "msg_1001"
  }
}

18.2 幂等

实时消息可能重复到达，前端和后端都要能去重。

前端：

typescript

const seen = new Set<string>();

function handleMessage(message: { id: string }) {
  if (seen.has(message.id)) return;
  seen.add(message.id);
  render(message);
}

后端：

消息有全局唯一 ID。
消费者处理前检查是否已处理。
数据库写入使用唯一约束。
外部调用使用幂等 key。

18.3 背压

背压是：生产速度大于消费速度时，系统如何不被压垮。

常见策略：

限制单连接发送速率。
合并高频消息，例如 100ms 合并一次指标。
丢弃可丢消息，例如鼠标轨迹。
队列设置上限。
慢客户端断开。
Worker 消费限流。
使用消息队列缓冲。

背压的三个层次（很多团队只做了第三层，问题出在前两层）：

text

层次一 网络/内核   TCP 窗口收缩，数据滞留内核，对端 send 变慢
层次二 运行时流    Node: write() 返回 false + highWaterMark（stream 默认 16KiB）
                  Web Streams: desiredSize ≤ 0 → 停止 pull
层次三 应用层      业务队列上限、合并、丢弃、断开慢客户端

WebSocket 发送前看缓冲：

javascript

if (socket.bufferedAmount > 1024 * 1024) {
  socket.close(1013, "client too slow");
}

浏览器 WebSocket.bufferedAmount 是已 send() 但尚未 transmitted 到网络的应用数据字节数（UTF-8 计），不含帧头开销、不含操作系统/网卡的缓冲；且按规范，连接关闭后它不会清零，只会随继续 send() 增长——用它做限速要同时检查 readyState：

javascript

async function sendThrottled(socket, data, maxQueued = 1024 * 1024) {
  while (socket.readyState === WebSocket.OPEN &&
         socket.bufferedAmount > maxQueued) {
    await new Promise((r) => setTimeout(r, 50)); // 轮询缓冲是 MDN 推荐做法
  }
  if (socket.readyState !== WebSocket.OPEN) throw new Error("socket not open");
  socket.send(data);
}

Node ws 库对应属性是 socket.writableLength，语义相同（等待写出的队列字节数）。

服务端“对每个消费者只保留最新值”的 conflation（合并）模式，是高频遥测/进度场景最实用的背压策略：

javascript

let latest = null;                 // 只留快照，不留历史
function publish(state) { latest = state; }
setInterval(() => {                // 固定节拍出队，节拍即限速器
  if (latest) { broadcast(latest); latest = null; }
}, 100);

对进度、看板这类“新值覆盖旧值”的语义是安全的；对聊天消息、告警这类不可丢事件，则改为“有界队列 + 满了先丢最旧可丢的 + 超过阈值断开客户端让其走续读”。

判断标准一句话：这条数据被新数据覆盖后，用户是否还需要看到它？ 需要 → 队列 + 持久化 + 断线续读；不需要 → 快照合并。

18.4 心跳

心跳解决两个问题：

发现半开连接。
防止中间代理长时间无数据断开。

WebSocket：

text

server -> ping
client -> pong

SSE：

text

: heartbeat

SSE 注释行以 : 开头，浏览器会忽略，但可以保持连接活跃。

19. 测试、监控与排查

19.1 SSE 测试

bash

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

-N 表示不缓冲输出。

检查：

响应头是不是 text/event-stream。
每条事件后有没有空行。
Nginx 是否关闭 buffering。
浏览器 Network 是否持续 pending。

19.2 WebSocket 测试

可以使用浏览器 DevTools Network 的 WS 面板，或命令行工具。

检查：

握手是否返回 101。
Upgrade 头是否正确。
是否有心跳。
断线是否重连。
多实例广播是否正常。

19.3 监控指标

实时系统至少监控：

当前连接数。
每秒消息数。
单连接发送队列长度。
重连次数。
平均延迟和 P95/P99 延迟。
SSE/WebSocket 错误数。
Broker lag。
消费失败和重试次数。
网关 499/502/504。
内存和文件描述符。

19.4 常见故障

现象	可能原因
SSE 本地正常，线上不实时	Nginx buffering、CDN 缓冲、网关超时
WebSocket 握手失败	Upgrade 头没透传、CORS、鉴权失败
多实例用户收不到消息	只用了本机内存 room，没有跨实例广播
页面越来越卡	消息无限累积、没有列表截断、没有虚拟滚动
服务内存涨	断开连接后没有清理 listener/timer/队列
AI 流式很慢	上游慢、代理缓冲、前端解码策略不对
重复消息	重连后没有按 messageId 去重

20. 学习路线与练习项目

20.1 学习路线

理解 HTTP 请求响应模型。
写短轮询任务进度。
用 SSE 改造任务进度。
用 fetch ReadableStream 做 AI 文本流。
写 WebSocket 聊天室。
给 WebSocket 加心跳、重连、鉴权。
用 Redis Pub/Sub 支持多实例广播。
用 Redis Streams 或 Kafka 做可重放事件流。
配置 Nginx SSE 和 WebSocket。
接入监控和压测。

20.2 练习项目一：AI Chat Stream

技术栈：

Vue 3 + Pinia。
NestJS 或 FastAPI。
SSE 或 HTTP Stream。
PostgreSQL 保存会话。

功能：

新建会话。
发送消息。
流式展示回答。
停止生成。
错误重试。
保存消息。

20.3 练习项目二：实时任务中心

技术栈：

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

功能：

创建导入任务。
Worker 消费任务。
推送进度和错误行。
支持断线续读。
生成错误报告。

20.4 练习项目三：实时通知中心

技术栈：

Vue 3 + Pinia。
NestJS Gateway 或 FastAPI WebSocket。
Redis Pub/Sub。

功能：

用户上线。
加入租户房间。
发布通知。
小红点更新。
多实例广播。
断线重连和消息去重。

21. 官方参考资料

MDN Server-Sent Events：https://developer.mozilla.org/en-US/docs/Web/API/Server-sent_events
MDN EventSource：https://developer.mozilla.org/en-US/docs/Web/API/EventSource
MDN WebSocket API：https://developer.mozilla.org/en-US/docs/Web/API/WebSocket
MDN Streams API：https://developer.mozilla.org/en-US/docs/Web/API/Streams_API
Node.js Streams：https://nodejs.org/api/stream.html
NestJS Server-Sent Events：https://docs.nestjs.com/techniques/server-sent-events
NestJS WebSockets：https://docs.nestjs.com/websockets/gateways
FastAPI WebSockets：https://fastapi.tiangolo.com/advanced/websockets/
FastAPI Custom Response / StreamingResponse：https://fastapi.tiangolo.com/advanced/custom-response/
Spring Framework WebSocket：https://docs.spring.io/spring-framework/reference/web/websocket.html
Spring WebFlux：https://docs.spring.io/spring-framework/reference/web/webflux.html
Vue Composition API：https://vuejs.org/api/composition-api-setup.html
Nginx WebSocket proxying：https://nginx.org/en/docs/http/websocket.html

流式处理与实时通信系统学习文档 ​

目录 ​

1. 为什么要学流式处理和实时通信 ​

2. 通信方式总览与选型表 ​

2.1 常见方式 ​

2.2 快速选型 ​

2.3 不要过度设计 ​

2.4 容易被忽略的选型变量：传输层协议版本 ​

3. 短轮询与长轮询 ​

3.1 短轮询 ​

3.2 长轮询 ​

3.3 内部机制：长轮询占用的到底是什么资源 ​

4. SSE：服务端事件推送 ​

4.1 SSE 数据格式 ​

4.2 浏览器 EventSource ​

4.3 SSE 适合 AI 流式输出 ​

4.4 注意事项：EventSource 的三条容易踩空的规范行为 ​

5. WebSocket：全双工实时通信 ​

5.1 浏览器 WebSocket ​

5.2 消息协议设计 ​

5.3 WebSocket 适合什么 ​

5.4 心跳与重连 ​

5.5 内部机制：握手与帧格式 ​

6. HTTP 流与 ReadableStream ​

6.1 fetch 读取流 ​

6.2 SSE vs HTTP Stream ​

6.3 内部机制：ReadableStream 的队列、水位线与背压 ​

6.4 注意事项：解码与切帧必须在消费侧做 ​

7. WebRTC、MQTT、gRPC Streaming 的定位 ​

7.1 WebRTC ​

7.2 MQTT ​

7.3 gRPC Streaming ​

8. 前端框架落地：Vue、Nuxt、React 思路 ​

8.1 Vue：封装 useSse ​

8.2 Vue：封装 useWebSocket ​

8.3 Pinia 管理实时状态 ​

8.4 Nuxt 落地注意点 ​

8.5 React 思路 ​

9. Node.js 与 Express 落地 ​

9.1 Express SSE ​

9.2 Express + HTTP Stream ​

9.3 ws WebSocket Server ​

10. NestJS 落地 ​

10.1 NestJS SSE ​

10.2 NestJS WebSocket Gateway ​

10.3 NestJS 鉴权 ​

11. Python FastAPI 落地 ​

11.1 StreamingResponse ​

11.2 FastAPI WebSocket ​

11.3 FastAPI + 队列推送任务进度 ​

12. Java Spring Boot 落地 ​

12.1 Spring MVC SseEmitter ​

12.2 Spring WebFlux SSE ​

12.3 Spring WebSocket / STOMP ​

13. 消息队列与流式数据处理 ​

13.1 Pub/Sub 与 Stream 的区别 ​

13.2 Redis Pub/Sub ​

13.3 Redis Streams ​

13.4 Kafka ​

13.5 RabbitMQ ​

14. 项目落地一：AI 流式聊天助手 ​

14.1 需求 ​

14.2 推荐架构 ​

14.3 为什么选 SSE ​

14.4 接口设计 ​

14.5 前端落地 ​

14.6 后端落地重点 ​

14.7 数据库表建议 ​

15. 项目落地二：实时通知和运营看板 ​

15.1 需求 ​

15.2 推荐架构 ​

15.3 WebSocket 房间设计 ​

15.4 前端 Pinia 接入 ​

15.5 后端多实例广播 ​

16. 项目落地三：任务进度、日志流和大文件处理 ​

16.1 需求 ​

16.2 推荐架构 ​

16.3 事件设计 ​

16.4 为什么不用 WebSocket ​

17. 部署、网关、鉴权与横向扩展 ​

流式处理与实时通信系统学习文档

目录

1. 为什么要学流式处理和实时通信

2. 通信方式总览与选型表

2.1 常见方式

2.2 快速选型

2.3 不要过度设计

2.4 容易被忽略的选型变量：传输层协议版本

3. 短轮询与长轮询

3.1 短轮询

3.2 长轮询

3.3 内部机制：长轮询占用的到底是什么资源

4. SSE：服务端事件推送

4.1 SSE 数据格式

4.2 浏览器 EventSource

4.3 SSE 适合 AI 流式输出

4.4 注意事项：EventSource 的三条容易踩空的规范行为

5. WebSocket：全双工实时通信

5.1 浏览器 WebSocket

5.2 消息协议设计

5.3 WebSocket 适合什么

5.4 心跳与重连

5.5 内部机制：握手与帧格式

6. HTTP 流与 ReadableStream

6.1 fetch 读取流

6.2 SSE vs HTTP Stream

6.3 内部机制：ReadableStream 的队列、水位线与背压

6.4 注意事项：解码与切帧必须在消费侧做

7. WebRTC、MQTT、gRPC Streaming 的定位

7.1 WebRTC

7.2 MQTT

7.3 gRPC Streaming

8. 前端框架落地：Vue、Nuxt、React 思路

8.1 Vue：封装 useSse

8.2 Vue：封装 useWebSocket

8.3 Pinia 管理实时状态

8.4 Nuxt 落地注意点

8.5 React 思路

9. Node.js 与 Express 落地

9.1 Express SSE

9.2 Express + HTTP Stream

9.3 ws WebSocket Server

10. NestJS 落地

10.1 NestJS SSE

10.2 NestJS WebSocket Gateway

10.3 NestJS 鉴权

11. Python FastAPI 落地

11.1 StreamingResponse

11.2 FastAPI WebSocket

11.3 FastAPI + 队列推送任务进度

12. Java Spring Boot 落地

12.1 Spring MVC SseEmitter

12.2 Spring WebFlux SSE

12.3 Spring WebSocket / STOMP

13. 消息队列与流式数据处理

13.1 Pub/Sub 与 Stream 的区别

13.2 Redis Pub/Sub

13.3 Redis Streams

13.4 Kafka

13.5 RabbitMQ

14. 项目落地一：AI 流式聊天助手

14.1 需求

14.2 推荐架构

14.3 为什么选 SSE

14.4 接口设计

14.5 前端落地

14.6 后端落地重点

14.7 数据库表建议

15. 项目落地二：实时通知和运营看板

15.1 需求

15.2 推荐架构

15.3 WebSocket 房间设计

15.4 前端 Pinia 接入

15.5 后端多实例广播

16. 项目落地三：任务进度、日志流和大文件处理

16.1 需求

16.2 推荐架构

16.3 事件设计

16.4 为什么不用 WebSocket

17. 部署、网关、鉴权与横向扩展