⚡ 实时通信

实时通信是什么？为什么 SaaS 离不开它

实时通信是指客户端与服务器（或客户端之间）在数据产生后立即传递的能力，而非依赖客户端定时轮询（Polling）。在现代 SaaS 产品中，实时性已经从"锦上添花"变成了"基础能力"。

如果不使用实时通信，替代方案是轮询（Polling）：客户端每隔 N 秒请求一次。问题显而易见：

传输层：WebSocket vs SSE vs Long Polling

三种主要的实时通信传输方式，各有适用场景：

WebSocket — 全双工持久连接

// 握手过程
客户端 → 服务器: HTTP GET /ws
  Upgrade: websocket
  Connection: Upgrade
  Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==

服务器 → 客户端: HTTP 101 Switching Protocols
  Upgrade: websocket
  Connection: Upgrade
  Sec-WebSocket-Accept: s3pPLMBiTxaQ9kYGzzhZRbK+xOo=

// 握手后：全双工通信
客户端 ←→ 服务器: 二进制帧或文本帧

Server-Sent Events (SSE) — 服务器单向推送

// 服务端 (Node.js)
res.writeHead(200, {
  'Content-Type': 'text/event-stream',
  'Cache-Control': 'no-cache',
  'Connection': 'keep-alive'
});

// 发送事件
res.write('event: message\n');
res.write('data: {"text":"hello"}\n');
res.write('id: 123\n\n');  // ID 用于自动重连续传

// 客户端 (浏览器)
const es = new EventSource('/sse');
es.addEventListener('message', (e) => {
  console.log(JSON.parse(e.data));
});
// 断线自动重连！浏览器原生支持

Long Polling — 兼容性最后的兜底

传输层对比总览

WebSocket 深度实战

服务端：Node.js + ws

import { WebSocketServer } from 'ws';

const wss = new WebSocketServer({ port: 8080 });

// 连接管理
const clients = new Map(); // ws → { userId, rooms: Set }

wss.on('connection', (ws, req) => {
  // 1. 认证：从 query 或 header 提取 token
  const token = new URL(req.url, 'http://localhost').searchParams.get('token');
  const user = verifyToken(token);
  if (!user) { ws.close(4001, 'Unauthorized'); return; }

  // 2. 注册
  clients.set(ws, { userId: user.id, rooms: new Set() });

  // 3. 心跳
  ws.isAlive = true;
  ws.on('pong', () => { ws.isAlive = true; });

  // 4. 消息处理
  ws.on('message', (raw) => {
    const msg = JSON.parse(raw);
    handleMessage(ws, msg);
  });

  ws.on('close', () => {
    clients.delete(ws);
    broadcastPresence(user.id, 'offline');
  });
});

// 心跳检测（30s 间隔）
const heartbeat = setInterval(() => {
  wss.clients.forEach((ws) => {
    if (!ws.isAlive) return ws.terminate();
    ws.isAlive = false;
    ws.ping();
  });
}, 30000);

wss.on('close', () => clearInterval(heartbeat));

// 消息路由
function handleMessage(ws, msg) {
  const client = clients.get(ws);
  switch (msg.type) {
    case 'join_room':
      client.rooms.add(msg.roomId);
      broadcastToRoom(msg.roomId, {
        type: 'user_joined',
        userId: client.userId
      });
      break;
    case 'chat':
      // 持久化 + 广播
      saveMessage(msg.roomId, client.userId, msg.text);
      broadcastToRoom(msg.roomId, {
        type: 'chat',
        userId: client.userId,
        text: msg.text,
        timestamp: Date.now()
      });
      break;
  }
}

// 房间广播
function broadcastToRoom(roomId, data) {
  const payload = JSON.stringify(data);
  clients.forEach((client, ws) => {
    if (client.rooms.has(roomId) && ws.readyState === 1) {
      ws.send(payload);
    }
  });
}

服务端：Python + FastAPI

from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from typing import Dict, Set

app = FastAPI()

class ConnectionManager:
    def __init__(self):
        self.active: Dict[str, Set[WebSocket]] = {}  # room → connections

    async def connect(self, ws: WebSocket, room: str):
        await ws.accept()
        if room not in self.active:
            self.active[room] = set()
        self.active[room].add(ws)

    def disconnect(self, ws: WebSocket, room: str):
        self.active[room].discard(ws)

    async def broadcast(self, room: str, message: dict):
        for ws in self.active.get(room, set()):
            await ws.send_json(message)

manager = ConnectionManager()

@app.websocket("/ws/{room}/{user_id}")
async def ws_endpoint(ws: WebSocket, room: str, user_id: str):
    await manager.connect(ws, room)
    await manager.broadcast(room, {"type": "join", "user": user_id})
    try:
        while True:
            data = await ws.receive_json()
            await manager.broadcast(room, {
                "type": "message",
                "user": user_id,
                "text": data["text"],
            })
    except WebSocketDisconnect:
        manager.disconnect(ws, room)
        await manager.broadcast(room, {"type": "leave", "user": user_id})

客户端：原生 + 重连逻辑

class RobustWebSocket {
  constructor(url, options = {}) {
    this.url = url;
    this.reconnectInterval = options.reconnectInterval || 1000;
    this.maxReconnectInterval = options.maxReconnectInterval || 30000;
    this.heartbeatInterval = options.heartbeatInterval || 30000;
    this.handlers = {};
    this.reconnectAttempts = 0;
    this.connect();
  }

  connect() {
    this.ws = new WebSocket(this.url);

    this.ws.onopen = () => {
      this.reconnectAttempts = 0;
      this.reconnectInterval = 1000;
      this.startHeartbeat();
      this.emit('open');
    };

    this.ws.onmessage = (e) => {
      const data = JSON.parse(e.data);
      this.emit('message', data);
      this.emit(data.type, data);  // 按类型分发
    };

    this.ws.onclose = (e) => {
      this.stopHeartbeat();
      if (e.code !== 1000) {  // 非正常关闭 → 重连
        this.scheduleReconnect();
      }
      this.emit('close', e);
    };

    this.ws.onerror = (e) => {
      this.emit('error', e);
    };
  }

  scheduleReconnect() {
    const delay = Math.min(
      this.reconnectInterval * Math.pow(1.5, this.reconnectAttempts),
      this.maxReconnectInterval
    );
    this.reconnectAttempts++;
    setTimeout(() => this.connect(), delay);
  }

  startHeartbeat() {
    this.heartbeatTimer = setInterval(() => {
      if (this.ws.readyState === WebSocket.OPEN) {
        this.ws.send(JSON.stringify({ type: 'ping' }));
      }
    }, this.heartbeatInterval);
  }

  stopHeartbeat() {
    clearInterval(this.heartbeatTimer);
  }

  send(data) {
    if (this.ws.readyState === WebSocket.OPEN) {
      this.ws.send(JSON.stringify(data));
    }
  }

  on(event, handler) {
    if (!this.handlers[event]) this.handlers[event] = [];
    this.handlers[event].push(handler);
  }

  emit(event, ...args) {
    (this.handlers[event] || []).forEach(h => h(...args));
  }
}

// 使用
const ws = new RobustWebSocket('ws://localhost:8080?token=xxx');
ws.on('chat', (data) => {
  appendMessage(data.userId, data.text);
});
ws.on('open', () => {
  ws.send({ type: 'join_room', roomId: 'general' });
});

Socket.IO — WebSocket 的生产级封装

// 服务端
import { Server } from 'socket.io';

const io = new Server(3000, {
  cors: { origin: '*' }
});

io.on('connection', (socket) => {
  console.log(`User connected: ${socket.id}`);

  // 加入房间
  socket.on('join', (roomId) => {
    socket.join(roomId);
    socket.to(roomId).emit('user_joined', { id: socket.id });
  });

  // 聊天消息
  socket.on('chat', (data) => {
    io.to(data.roomId).emit('chat', {
      userId: socket.id,
      text: data.text,
      timestamp: Date.now()
    });
  });

  // 断线
  socket.on('disconnect', (reason) => {
    console.log(`Disconnected: ${socket.id} (${reason})`);
  });
});

// 向特定房间广播
io.to('room-123').emit('notification', { text: 'Hello room!' });

Server-Sent Events 实战

SSE 最适合的场景是服务器单向推送——通知、流式 AI、实时指标。

SSE + LLM 流式输出

// Next.js API Route: 流式返回 LLM 生成内容
export async function POST(req: Request) {
  const { prompt } = await req.json();

  const stream = await openai.chat.completions.create({
    model: 'gpt-4o',
    messages: [{ role: 'user', content: prompt }],
    stream: true,
  });

  const encoder = new TextEncoder();
  const readable = new ReadableStream({
    async start(controller) {
      for await (const chunk of stream) {
        const content = chunk.choices[0]?.delta?.content || '';
        if (content) {
          // SSE 格式
          controller.enqueue(
            encoder.encode(`data: ${JSON.stringify({ content })}\n\n`)
          );
        }
      }
      controller.enqueue(encoder.encode('data: [DONE]\n\n'));
      controller.close();
    },
  });

  return new Response(readable, {
    headers: {
      'Content-Type': 'text/event-stream',
      'Cache-Control': 'no-cache',
      'Connection': 'keep-alive',
    },
  });
}

// 客户端消费
const response = await fetch('/api/chat', {
  method: 'POST',
  body: JSON.stringify({ prompt }),
});
const reader = response.body.getReader();
const decoder = new TextDecoder();

while (true) {
  const { done, value } = await reader.read();
  if (done) break;
  const text = decoder.decode(value);
  // 解析 SSE data: 行
  const lines = text.split('\n').filter(l => l.startsWith('data: '));
  for (const line of lines) {
    const data = line.slice(6);
    if (data === '[DONE]') break;
    const { content } = JSON.parse(data);
    appendToUI(content);  // 逐字追加到 UI
  }
}

SSE + Vercel AI SDK

// app/api/chat/route.ts (Next.js App Router)
import { streamText } from 'ai';
import { openai } from '@ai-sdk/openai';

export async function POST(req: Request) {
  const { messages } = await req.json();
  const result = streamText({ model: openai('gpt-4o'), messages });
  return result.toDataStreamResponse();  // 自动 SSE 格式
}

// 客户端
'use client';
import { useChat } from 'ai/react';

export default function Chat() {
  const { messages, input, handleInputChange, handleSubmit } = useChat();
  return (
    <div>
      {messages.map(m => (
        <div key={m.id}>{m.role}: {m.content}</div>
      ))}
      <form onSubmit={handleSubmit}>
        <input value={input} onChange={handleInputChange} />
      </form>
    </div>
  );
}

SSE 用于实时通知推送

// 服务端：推送通知
export function createNotificationStream(userId: string) {
  const encoder = new TextEncoder();

  return new ReadableStream({
    start(controller) {
      // 监听 Redis Pub/Sub
      const sub = redis.subscribe(`notifications:${userId}`);
      sub.on('message', (channel, message) => {
        controller.enqueue(
          encoder.encode(`event: notification\ndata: ${message}\n\n`)
        );
      });

      // 心跳（防止连接超时）
      const heartbeat = setInterval(() => {
        controller.enqueue(encoder.encode(':heartbeat\n\n'));
      }, 15000);

      // 清理
      controller.enqueue = (...args) => {
        if (controller.desiredSize === null) {
          clearInterval(heartbeat);
          sub.unsubscribe();
          return;
        }
        return ReadableStream.prototype.enqueue.apply(controller, args);
      };
    }
  });
}

CRDT：实时协作的核心算法

CRDT（Conflict-free Replicated Data Types，无冲突复制数据类型）是实时协作的数学基础。它解决了核心问题：多人同时编辑同一份数据，如何保证最终一致且不丢失信息。

为什么需要 CRDT？

CRDT 的核心原理

CRDT 数据类型

CRDT vs OT

Yjs — 最流行的 CRDT 实现

核心概念

import * as Y from 'yjs';

// 1. 创建文档
const doc = new Y.Doc();

// 2. 使用 Shared Types（和普通 JS 类型一样操作）
const ytext = doc.getText('content');
const ymap = doc.getMap('metadata');
const yarray = doc.getArray('items');

// 3. 操作 — 和原生类型几乎一样
ytext.insert(0, 'Hello ');
ytext.insert(6, 'World');

ymap.set('title', 'My Doc');
ymap.set('version', 1);

yarray.push([{ name: 'Item 1' }]);
yarray.insert(0, [{ name: 'Item 0' }]);

// 4. 监听变化
ytext.observe(event => {
  console.log('Text changed:', ytext.toString());
  event.delta.forEach(change => {
    // { insert: string } | { delete: number } | { retain: number }
  });
});

多端同步

import * as Y from 'yjs';
import { WebsocketProvider } from 'y-websocket';

const doc = new Y.Doc();

// 连接到 WebSocket 服务器
const provider = new WebsocketProvider(
  'wss://your-server.com',
  'room-name',
  doc
);

// 同步状态
provider.on('status', event => {
  console.log(event.status); // 'connecting' | 'connected' | 'disconnected'
});

// 自动同步！编辑 ytext → 其他客户端自动收到
const ytext = doc.getText('content');
ytext.insert(0, 'Hello from client A!');

Yjs + React

import { useYjs } from 'react-yjs';
import * as Y from 'yjs';

function CollaborativeEditor({ roomId }) {
  const doc = useYjs(roomId);
  const ytext = doc.getText('content');

  return (
    <textarea
      value={ytext.toString()}
      onChange={(e) => {
        // 简化版：实际需要用 ytext.delete + ytext.insert 处理差异
        doc.transact(() => {
          ytext.delete(0, ytext.length);
          ytext.insert(0, e.target.value);
        });
      }}
    />
  );
}

Yjs 服务端

// 最简 Yjs WebSocket 服务器
import { Server } from 'ws';
import * as Y from 'yjs';
import { setupWSConnection } from 'y-websocket/bin/utils';

const wss = new Server({ port: 1234 });

const docs = new Map(); // room → Y.Doc

wss.on('connection', (ws, req) => {
  const room = new URL(req.url, 'http://localhost').searchParams.get('room');
  if (!docs.has(room)) docs.set(room, new Y.Doc());
  const doc = docs.get(room);

  setupWSConnection(ws, req, { doc });
  // y-websocket 自带：同步、感知、更新广播
});

Automerge — 数据版本控制

import * as Automerge from '@automerge/automerge';

// 1. 创建文档
let doc = Automerge.init();
doc = Automerge.change(doc, d => {
  d.title = 'Meeting Notes';
  d.items = [];
  d.items.push({ text: 'Discuss roadmap', done: false });
});

// 2. 修改
doc = Automerge.change(doc, d => {
  d.items[0].done = true;
  d.items.push({ text: 'Review PRs', done: false });
});

// 3. 获取变更（类似 git diff）
const changes = Automerge.getChanges(prevDoc, doc);

// 4. 应用远端变更（类似 git merge）
doc = Automerge.applyChanges(doc, remoteChanges);

// 5. 查看历史
const history = Automerge.getHistory(doc);
// [{ change: { ... }, snapshot: { title: 'Meeting Notes', items: [...] } }, ...]

// 6. Fork + Merge（分支开发）
let fork = Automerge.fork(doc);
fork = Automerge.change(fork, d => { d.title = 'Updated Title'; });
doc = Automerge.merge(doc, fork);  // 合并回来

Liveblocks — 托管式实时协作

// Next.js + Liveblocks
import { createClient } from '@liveblocks/client';
import { LiveblocksProvider, RoomProvider } from '@liveblocks/react';

const client = createClient({
  publicApiKey: 'pk_live_xxx',
});

function App() {
  return (
    <LiveblocksProvider client={client}>
      <RoomProvider id="my-room">
        <CollaborativeEditor />
      </RoomProvider>
    </LiveblocksProvider>
  );
}

// 使用 Storage（CRDT 数据）
import { useStorage, useSelf, useOthers } from '@liveblocks/react';

function CollaborativeEditor() {
  // CRDT 数据
  const text = useStorage(root => root.document.textContent);

  // 自己的信息
  const me = useSelf();
  // 其他人的信息
  const others = useOthers();

  return (
    <div>
      <div className="presence-bar">
        {others.map(user => (
          <Avatar key={user.id} color={user.info.color} />
        ))}
      </div>
      {/* 编辑器内容 */}
    </div>
  );
}

Presence（在线状态）系统

在线状态是实时应用中"看得见对方"的关键——光标位置、输入状态、头像颜色。它和 CRDT 数据不同：状态是短暂的、不需要持久化、最新值覆盖旧值。

// Yjs Awareness 协议（y-protocols 自带）
import { awarenessProtocol } from 'y-protocols/awareness';

const awareness = new awarenessProtocol.Awareness(doc);

// 设置自己的状态
awareness.setLocalState({
  name: 'Alice',
  color: '#ff6b6b',
  cursor: { line: 10, col: 5 },
  selection: { from: 45, to: 52 },
  isTyping: true,
});

// 监听他人的状态
awareness.on('change', ({ added, updated, removed }) => {
  const states = awareness.getStates();
  // Map(clientId → { name, color, cursor, ... })
});

// 超时自动移除（默认 30s 无心跳）

扩展性：从 100 到 100 万连接

阶段 1：单服务器（0 → 5,000 连接）

// 单进程 Node.js + ws/Socket.IO
// 单机约 5K-10K 连接（取决于消息频率）
const wss = new WebSocketServer({ port: 8080 });
// 直接进程内 Map 管理房间

阶段 2：多进程（5K → 50,000 连接）

// 使用 Redis Pub/Sub 做进程间通信
import { createClient } from 'redis';

const sub = createClient();
const pub = createClient();
await sub.connect();
await pub.connect();

// 订阅房间消息
sub.subscribe('room:123', (message) => {
  // 广播给本进程的连接
  broadcastToLocal('room:123', JSON.parse(message));
});

// 发送消息到房间
function publishToRoom(roomId, data) {
  pub.publish(`room:${roomId}`, JSON.stringify(data));
}

// 配合 pm2 cluster 或 Node.js cluster

阶段 3：多服务器（50K → 500,000+ 连接）

// Nginx Sticky Session 配置
upstream websocket {
  ip_hash;  // 同一 IP → 同一后端
  server ws1:8080;
  server ws2:8080;
  server ws3:8080;
}

server {
  location /ws {
    proxy_pass http://websocket;
    proxy_http_version 1.1;
    proxy_set_header Upgrade $http_upgrade;
    proxy_set_header Connection "upgrade";
    proxy_read_timeout 86400s;  // 24h 超时
  }
}

阶段 4：大规模（500K+ → 百万级）

Agent 系统中的实时通信

从我们的 Agent 架构研究 中，实时通信在 Agent 系统中扮演关键角色：

Agent → 用户的实时流

// Agent 执行过程的实时推送
// 使用 SSE（单向够用，Agent → 用户）

// 服务端
export async function POST(req: Request) {
  const { task } = await req.json();

  return new Response(
    new ReadableStream({
      async start(controller) {
        const encoder = new TextEncoder();

        function send(event: string, data: any) {
          controller.enqueue(
            encoder.encode(`event: ${event}\ndata: ${JSON.stringify(data)}\n\n`)
          );
        }

        // 1. 开始思考
        send('thinking', { step: 'analyzing', message: 'Analyzing task...' });

        // 2. 工具调用
        send('tool_call', { tool: 'web_search', args: { query: '...' } });

        // 3. 工具结果
        send('tool_result', { tool: 'web_search', result: '...' });

        // 4. 需要审批
        send('approval_needed', {
          action: 'exec',
          command: 'rm -rf /tmp/old-data',
          risk: 'high'
        });

        // 5. 等待用户审批...
        // （审批通过另一个 HTTP 端点提交）

        // 6. 完成
        send('done', { result: 'Task completed successfully' });
        controller.close();
      }
    }),
    {
      headers: {
        'Content-Type': 'text/event-stream',
        'Cache-Control': 'no-cache',
      },
    }
  );
}

多 Agent 间的实时消息

// 使用 Redis Pub/Sub 做跨 Agent 通信
// 类似 AutoGen 的 Actor Model

class AgentBus {
  private redis: Redis;
  private agentId: string;

  async publish(targetAgent: string, message: AgentMessage) {
    await this.redis.publish(
      `agent:${targetAgent}`,
      JSON.stringify({ from: this.agentId, ...message })
    );
  }

  async subscribe(handler: (msg: AgentMessage) => void) {
    const sub = this.redis.duplicate();
    await sub.subscribe(`agent:${this.agentId}`, (raw) => {
      handler(JSON.parse(raw));
    });
  }

  // 广播给所有 Agent
  async broadcast(message: AgentMessage) {
    await this.redis.publish('agent:broadcast', JSON.stringify(message));
  }
}

选型决策树

替代方案与补充工具

相关链接