实时协作编辑前端的架构复盘：WebSocket 同步、冲突解决与离线支持

实时协作编辑前端的架构复盘：WebSocket 同步、冲突解决与离线支持

一、实时协作的三个核心挑战

实时协作编辑（如 Google Docs、Figma、Notion）的前端架构远比直觉中复杂。表面上只需要 WebSocket 推送变更、前端渲染更新；实际落地时需要同时解决三个技术难题：同步延迟与一致性——多个用户同时编辑同一段落时如何保证最终一致；冲突解决策略——OT（Operational Transformation）还是 CRDT（Conflict-free Replicated Data Type）；离线编辑与重连恢复——断网期间的本地编辑如何在恢复后无缝合并。

在一款文档协作产品的架构演进过程中，WebSocket 连接数从初期的单房间 50 人扩展至 500+ 人时，遇到了丢帧、乱序和重连风暴三类典型故障。

二、WebSocket 层的连接与重连策略

2.1 连接管理

WebSocket 连接管理涉及三个关键设计：心跳保活、指数退避重连、连接去重。在生产环境中观察到的最常见问题是：当服务端因滚动发布断开连接时，所有客户端几乎同一时刻检测到断线并同时发起重连，造成"重连风暴"。

// websocket-manager.ts — WebSocket 连接管理器

/** 连接状态 */
type ConnectionState = 'connecting' | 'connected' | 'reconnecting' | 'disconnected';

/** 重连策略配置 */
interface ReconnectConfig {
/** 初始重连延迟（毫秒） */
initialDelay: number;
/** 最大重连延迟（毫秒） */
maxDelay: number;
/** 延迟增长因子（指数退避基数） */
backoffMultiplier: number;
/** 添加随机抖动的范围（百分比，0-1） */
jitterPercent: number;
/** 最大重连次数（超过后放弃） */
maxAttempts: number;
}

/**
* WebSocket 连接管理器
* 负责连接生命周期管理、心跳保活、指数退避重连
*/
export class WebSocketManager {
private ws: WebSocket | null = null;
private url: string;
private reconnectConfig: ReconnectConfig;
private reconnectAttempt = 0;
private reconnectTimer: ReturnType<typeof setTimeout> | null = null;
private heartbeatTimer: ReturnType<typeof setInterval> | null = null;
private state: ConnectionState = 'disconnected';

/** 状态变化回调 */
private stateCallbacks: Set<(state: ConnectionState) => void> = new Set();
/** 消息接收回调 */
private messageCallbacks: Set<(data: unknown) => void> = new Set();

constructor(url: string, config?: Partial<ReconnectConfig>) {
this.url = url;
this.reconnectConfig = {
initialDelay: 1000,
maxDelay: 30000,
backoffMultiplier: 2,
jitterPercent: 0.3,
maxAttempts: 10,
…config,
};
}

/** 建立连接 */
connect(): void {
if (this.state === 'connected' || this.state === 'connecting') return;
this.setState('connecting');

try {
this.ws = new WebSocket(this.url);
this.ws.onopen = () => this.handleOpen();
this.ws.onmessage = (event) => this.handleMessage(event);
this.ws.onclose = (event) => this.handleClose(event);
this.ws.onerror = () => this.handleError();
} catch (err) {
console.error('[WS Manager] 创建连接失败:', err);
this.scheduleReconnect();
}
}

/** 发送消息（自动序列化） */
send(data: unknown): boolean {
if (this.state !== 'connected' || !this.ws) {
console.warn('[WS Manager] 连接未建立，消息暂存到离线队列');
// 实际项目中应推入离线队列（IndexedDB）
return false;
}
try {
this.ws.send(JSON.stringify(data));
return true;
} catch (err) {
console.error('[WS Manager] 发送消息失败:', err);
return false;
}
}

/** 主动断开连接 */
disconnect(): void {
this.clearTimers();
if (this.ws) {
this.ws.onclose = null; // 阻止触发自动重连
this.ws.close(1000, '客户端主动断开');
this.ws = null;
}
this.setState('disconnected');
}

// ====== 私有方法 ======

private handleOpen(): void {
this.reconnectAttempt = 0;
this.setState('connected');
this.startHeartbeat();
}

private handleMessage(event: MessageEvent): void {
try {
const data = JSON.parse(event.data as string);
// 心跳响应不触发业务回调
if (data.type === 'pong') return;
for (const cb of this.messageCallbacks) {
try { cb(data); } catch { /* 隔离回调异常 */ }
}
} catch {
console.error('[WS Manager] 消息解析失败:', event.data);
}
}

/** 连接关闭处理：区分正常关闭与异常断线 */
private handleClose(event: CloseEvent): void {
this.clearTimers();
// 正常关闭（code 1000 为正常，1001 为页面离开）不触发重连
if (event.code === 1000 || event.code === 1001) {
this.setState('disconnected');
return;
}
this.scheduleReconnect();
}

private handleError(): void {
// onerror 之后必定触发 onclose，此处仅做日志记录
console.warn('[WS Manager] WebSocket 出错，等待 onclose 触发重连');
}

/** 调度重连（指数退避 + 随机抖动） */
private scheduleReconnect(): void {
if (this.reconnectAttempt >= this.reconnectConfig.maxAttempts) {
console.error(`[WS Manager] 已达最大重连次数（${this.reconnectConfig.maxAttempts}），放弃重连`);
this.setState('disconnected');
return;
}

this.setState('reconnecting');

// 计算延迟：指数退避
const baseDelay = Math.min(
this.reconnectConfig.initialDelay *
Math.pow(this.reconnectConfig.backoffMultiplier, this.reconnectAttempt),
this.reconnectConfig.maxDelay
);

// 添加随机抖动（避免重连风暴）
const jitter = baseDelay * this.reconnectConfig.jitterPercent * Math.random();
const delay = Math.round(baseDelay + jitter);

this.reconnectAttempt += 1;
this.reconnectTimer = setTimeout(() => {
this.reconnectTimer = null;
this.connect();
}, delay);
}

/** 心跳保活 */
private startHeartbeat(): void {
this.heartbeatTimer = setInterval(() => {
if (this.state === 'connected' && this.ws?.readyState === WebSocket.OPEN) {
this.ws.send(JSON.stringify({ type: 'ping', ts: Date.now() }));
} else {
// 心跳发送失败，强制触发重连
this.ws?.close();
}
}, 15000); // 15 秒心跳间隔
}

private setState(newState: ConnectionState): void {
this.state = newState;
for (const cb of this.stateCallbacks) {
try { cb(newState); } catch { /* 隔离 */ }
}
}

private clearTimers(): void {
if (this.reconnectTimer) { clearTimeout(this.reconnectTimer); this.reconnectTimer = null; }
if (this.heartbeatTimer) { clearInterval(this.heartbeatTimer); this.heartbeatTimer = null; }
}
}

三、冲突解决：OT 与 CRDT 的工程取舍

在技术选型时决定采用 CRDT 而非 OT，基于以下三点判断：

// collaborative-editor.ts — 基于 Yjs 的协作编辑器封装
import * as Y from 'yjs';
import { WebsocketProvider } from 'y-websocket';

interface CollaborationConfig {
/** 房间 ID（文档唯一标识） */
roomId: string;
/** WebSocket 服务端地址 */
serverUrl: string;
/** 当前用户信息（用于感知协议） */
user: {
id: string;
name: string;
color: string;
};
/** 离线数据持久化回调 */
onPersist?: (update: Uint8Array) => Promise<void>;
/** 离线数据恢复回调 */
onRestore?: () => Promise<Uint8Array | null>;
/** 连接状态变化回调 */
onConnectionChange?: (connected: boolean) => void;
}

/**
* 协作编辑器核心封装
* 整合 CRDT 文档、WebSocket 同步与离线支持
*/
export class CollaborativeEditor {
private ydoc: Y.Doc;
private provider: WebsocketProvider;
private config: CollaborationConfig;
private _isConnected = false;

constructor(config: CollaborationConfig) {
this.config = config;

// 创建 Yjs 文档实例
this.ydoc = new Y.Doc();

// 建立 WebSocket 连接（y-websocket 内部处理重连）
this.provider = new WebsocketProvider(
config.serverUrl,
config.roomId,
this.ydoc
);

this.provider.on('status', ({ status }: { status: string }) => {
this._isConnected = status === 'connected';
config.onConnectionChange?.(this._isConnected);
});

// 设置用户感知信息（光标同步）
this.provider.awareness.setLocalState({
user: config.user,
});

// 监听文档更新用于离线持久化
this.ydoc.on('update', (update: Uint8Array) => {
config.onPersist?.(update);
});
}

/** 获取共享文本/富文本类型 */
getText(name: string = 'content'): Y.Text {
return this.ydoc.getText(name);
}

/** 获取共享 Map 类型（用于结构化数据同步） */
getMap<T = unknown>(name: string): Y.Map<T> {
return this.ydoc.getMap<T>(name);
}

/** 获取共享 Array 类型 */
getArray<T = unknown>(name: string): Y.Array<T> {
return this.ydoc.getArray<T>(name);
}

/** 检查连接状态 */
get isConnected(): boolean {
return this._isConnected;
}

/** 销毁并清理资源 */
destroy(): void {
this.provider.disconnect();
this.ydoc.destroy();
}
}

四、离线编辑的工程实现

离线编辑的核心流程是：用户操作 → 生成 CRDT 操作（本地 Yjs 文档更新） → 序列化变更（Y.encodeStateAsUpdate） → 持久化至 IndexedDB → 网络恢复后回放增量变更。

// offline-manager.ts — 离线编辑管理器
interface OfflineStore {
/** 存储增量更新 */
saveUpdate(docId: string, update: Uint8Array): Promise<void>;
/** 获取所有增量更新（按时间排序） */
getUpdates(docId: string): Promise<Uint8Array[]>;
/** 清除某个文档的离线数据 */
clear(docId: string): Promise<void>;
}

/**
* 基于 IndexedDB 的离线存储实现
*/
class IndexedDBOfflineStore implements OfflineStore {
private dbName = 'collaborative-editor-offline';
private storeName = 'updates';
private db: IDBDatabase | null = null;

private async getDB(): Promise<IDBDatabase> {
if (this.db) return this.db;
return new Promise((resolve, reject) => {
const request = indexedDB.open(this.dbName, 1);
request.onupgradeneeded = () => {
const db = request.result;
if (!db.objectStoreNames.contains(this.storeName)) {
const store = db.createObjectStore(this.storeName, {
keyPath: 'id',
autoIncrement: true,
} as IDBObjectStoreParameters);
store.createIndex('docId', 'docId', { unique: false });
}
};
request.onsuccess = () => {
this.db = request.result;
resolve(this.db);
};
request.onerror = () => reject(request.error);
});
}

async saveUpdate(docId: string, update: Uint8Array): Promise<void> {
const db = await this.getDB();
return new Promise((resolve, reject) => {
const tx = db.transaction(this.storeName, 'readwrite');
const store = tx.objectStore(this.storeName);
store.add({ docId, update, timestamp: Date.now() });
tx.oncomplete = () => resolve();
tx.onerror = () => reject(tx.error);
});
}

async getUpdates(docId: string): Promise<Uint8Array[]> {
const db = await this.getDB();
return new Promise((resolve, reject) => {
const tx = db.transaction(this.storeName, 'readonly');
const store = tx.objectStore(this.storeName);
const index = store.index('docId');
const request = index.getAll(IDBKeyRange.only(docId));
request.onsuccess = () => {
const records = request.result as Array<{
docId: string; update: Uint8Array; timestamp: number;
}>;
records.sort((a, b) => a.timestamp – b.timestamp);
resolve(records.map(r => r.update));
};
request.onerror = () => reject(request.error);
});
}

async clear(docId: string): Promise<void> {
const db = await this.getDB();
return new Promise((resolve, reject) => {
const tx = db.transaction(this.storeName, 'readwrite');
const store = tx.objectStore(this.storeName);
const index = store.index('docId');
const request = index.openCursor(IDBKeyRange.only(docId));
request.onsuccess = () => {
const cursor = request.result;
if (cursor) { cursor.delete(); cursor.continue(); }
};
tx.oncomplete = () => resolve();
tx.onerror = () => reject(tx.error);
});
}
}

export const offlineStore = new IndexedDBOfflineStore();

五、总结

实时协作前端的架构复杂性集中在三个环节：传输层（WebSocket 连接管理、心跳保活、防重连风暴）、数据层（CRDT/OT 的选择与集成）、离线层（增量持久化与恢复）。

在实践中得到的经验是：不要从零实现协作算法——Yjs 或 ShareDB 这样的成熟库已经解决了 90% 的分布式一致性问题，团队应该将精力集中在业务层的协作体验（如光标同步、评论定位、权限控制的 UI 反馈）和离线策略的调优（如 IndexedDB 的存储配额管理、清理过期数据）。

另外，WebSocket 连接数从 50 扩展到 500+ 时，需要服务端做消息批处理（coalescing）——将同一帧内的多次编辑操作合并为一次广播，避免带宽占用线性增长。