implementing-realtime-sync

实时同步

为应用实现实时通信，以支持实时更新、协作和在线状态感知。

适用场景

在以下场景中构建功能时使用此技能：

• 大模型流式界面 - 逐步流式传输 token（ai-chat 集成）
• 实时仪表盘 - 向客户端推送指标和更新
• 协作编辑 - 使用 CRDT 实现多用户文档／电子表格编辑
• 聊天应用 - 支持在线状态的实时消息传递
• 多人游戏功能 - 光标跟踪、实时更新和在线状态感知
• 离线优先应用 - 支持重新连接后同步的移动应用／PWA

协议选择框架

根据通信模式选择传输协议：

决策树

ONE-WAY (Server → Client only)
├─ LLM streaming, notifications, live feeds
└─ Use SSE (Server-Sent Events)
   ├─ Automatic reconnection (browser-native)
   ├─ Event IDs for resumption
   └─ Simple HTTP implementation

BIDIRECTIONAL (Client ↔ Server)
├─ Chat, games, collaborative editing
└─ Use WebSocket
   ├─ Manual reconnection required
   ├─ Binary + text support
   └─ Lower latency for two-way

COLLABORATIVE EDITING
├─ Multi-user documents/spreadsheets
└─ Use WebSocket + CRDT (Yjs or Automerge)
   ├─ CRDT handles conflict resolution
   ├─ WebSocket for transport
   └─ Offline-first with sync

PEER-TO-PEER MEDIA
├─ Video, screen sharing, voice calls
└─ Use WebRTC
   ├─ WebSocket for signaling
   ├─ Direct P2P connection
   └─ STUN/TURN for NAT traversal

协议对比

协议	方向	重新连接	复杂度	最适用场景
SSE	服务端 → 客户端	自动	低	实时动态、LLM 流式传输
WebSocket	双向	手动	中	聊天、游戏、协作
WebRTC	P2P	复杂	高	视频、屏幕共享、语音

实现模式

模式 1：使用 SSE 实现 LLM 流式传输

将 LLM token 逐步流式传输到前端（ai-chat 集成）。

Python（FastAPI）：

from sse_starlette.sse import EventSourceResponse

@app.post("/chat/stream")
async def stream_chat(prompt: str):
    async def generate():
        async for chunk in llm_stream:
            yield {"event": "token", "data": chunk.content}
        yield {"event": "done", "data": "[DONE]"}
    return EventSourceResponse(generate())

前端：

const es = new EventSource('/chat/stream')
es.addEventListener('token', (e) => appendToken(e.data))

完整实现、重新连接和事件 ID 恢复，请参阅 references/sse.md。

模式 2：WebSocket 聊天

为聊天应用实现双向通信。

Python（FastAPI）：

connections: set[WebSocket] = set()

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    connections.add(websocket)
    try:
        while True:
            data = await websocket.receive_text()
            for conn in connections:
                await conn.send_text(data)
    except WebSocketDisconnect:
        connections.remove(websocket)

多语言示例、身份验证、心跳和扩展，请参阅 references/websockets.md。

模式 3：使用 CRDT 实现协作编辑

使用 Yjs 实现无冲突的多用户编辑。

TypeScript（Yjs）：

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

const doc = new Y.Doc()
const provider = new WebsocketProvider('ws://localhost:1234', 'doc-id', doc)
const ytext = doc.getText('content')

ytext.observe(event => console.log('Changes:', event.changes))
ytext.insert(0, 'Hello collaborative world!')

冲突解决、Yjs 与 Automerge 的比较以及高级模式，请参阅 references/crdts.md。

模式 4：在线状态感知

跟踪在线用户、光标位置和输入指示器。

Yjs Awareness API：

const awareness = provider.awareness
awareness.setLocalState({ user: { name: 'Alice' }, cursor: { x: 100, y: 200 } })
awareness.on('change', () => {
  awareness.getStates().forEach((state, clientId) => {
    renderCursor(state.cursor, state.user)
  })
})

光标跟踪、输入指示器和在线状态，请参阅 references/presence-patterns.md。

模式 5：离线同步（移动应用／PWA）

在本地将变更加入队列，并在连接恢复后进行同步。

TypeScript（Yjs + IndexedDB）：

import { IndexeddbPersistence } from 'y-indexeddb'
import { WebsocketProvider } from 'y-websocket'

const doc = new Y.Doc()
const indexeddbProvider = new IndexeddbPersistence('my-doc', doc)
const wsProvider = new WebsocketProvider('wss://api.example.com/sync', 'my-doc', doc)

wsProvider.on('status', (e) => {
  console.log(e.status === 'connected' ? 'Online' : 'Offline')
})

冲突解决和同步策略，请参阅 references/offline-sync.md。

库推荐

Python

WebSocket：

• websockets 13.x - 基于 AsyncIO，可用于生产环境
• FastAPI WebSocket - 内置支持、依赖注入
• Flask-SocketIO - 支持回退机制的 Socket.IO 协议

SSE：

• sse-starlette - 适用于 FastAPI/Starlette，异步，基于生成器
• Flask-SSE - 用于发布／订阅的 Redis 后端

Rust

WebSocket：

• tokio-tungstenite 0.23 - 集成 Tokio，可用于生产环境
• axum WebSocket - 内置提取器、tower 中间件

SSE：

• axum SSE - 原生支持，异步流

Go

WebSocket：

• gorilla/websocket - 经过充分验证，支持压缩
• nhooyr/websocket - 现代 API，支持 context

SSE：

• net/http（原生）- Flusher 接口，无需依赖

TypeScript

WebSocket：

• ws - 原生 WebSocket 服务端，轻量
• Socket.io 4.x - 自动重新连接、回退机制、房间
• Hono WebSocket - 边缘运行时（Cloudflare Workers、Deno）

SSE：

• EventSource（原生）- 浏览器原生支持，自动重试
• Node.js http（原生）- 服务端使用，无需依赖

CRDT：

• Yjs - 成熟，支持 TypeScript/Rust，支持富文本编辑
• Automerge - Rust/JS，类似 JSON 的数据，支持时间回溯

重新连接策略

SSE： 浏览器的 EventSource 会使用指数退避自动处理重新连接。 WebSocket： 实现带抖动的手动指数退避，以防止惊群效应。

完整实现模式请参阅 references/sse.md 和 references/websockets.md。

安全模式

身份验证： 使用基于 Cookie 的身份验证（同源），或在 Sec-WebSocket-Protocol 标头中使用令牌。 速率限制： 使用滑动窗口实现按用户限流消息。

请参阅 references/websockets.md，了解身份验证和速率限制的实现方法。

使用 Redis Pub/Sub 进行扩展

如需水平扩展，可使用 Redis pub/sub 在多个后端服务之间广播消息。

请参阅 references/websockets.md，了解完整的 Redis 扩展实现。

前端集成

React Hooks 模式

用于 LLM 流式传输的 SSE（ai-chat）：

useEffect(() => {
  const es = new EventSource(`/api/chat/stream?prompt=${prompt}`)
  es.addEventListener('token', (e) => setContent(prev => prev + e.data))
  return () => es.close()
}, [prompt])

用于实时指标的 WebSocket（dashboards）：

useEffect(() => {
  const ws = new WebSocket('ws://localhost:8000/metrics')
  ws.onmessage = (e) => setMetrics(JSON.parse(e.data))
  return () => ws.close()
}, [])

用于协作表格的 Yjs：

useEffect(() => {
  const doc = new Y.Doc()
  const provider = new WebsocketProvider('ws://localhost:1234', docId, doc)
  const yarray = doc.getArray('rows')
  yarray.observe(() => setRows(yarray.toArray()))
  return () => provider.destroy()
}, [docId])

参考文档

如需详细了解实现模式，请参阅：

• references/sse.md - SSE 协议、重新连接、事件 ID
• references/websockets.md - WebSocket 身份验证、心跳、扩展
• references/crdts.md - Yjs 与 Automerge、冲突解决
• references/presence-patterns.md - 光标跟踪、输入指示器
• references/offline-sync.md - 移动端模式、冲突处理策略

示例项目

可用的实现示例：

• examples/llm-streaming-sse/ - 用于 LLM 流式传输的 FastAPI SSE（可运行）
• examples/chat-websocket/ - Python FastAPI + TypeScript 聊天
• examples/collaborative-yjs/ - Yjs 协作编辑器

测试工具

使用脚本验证实现：

• scripts/test_websocket_connection.py - WebSocket 连接测试