构建实时协作系统：从OT到CRDT的深度解析

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75

// LWW-Register (Last-Write-Wins Register)

class LWWRegister {
  constructor() {
    this.value = null;
    this.timestamp = 0;
  }

  set(value, timestamp = Date.now()) {
    if (timestamp >= this.timestamp) {
      this.value = value;
      this.timestamp = timestamp;
    }
    return this.value;
  }

  get() {
    return this.value;
  }

  merge(other) {
    if (other.timestamp > this.timestamp) {
      this.value = other.value;
      this.timestamp = other.timestamp;
    }
  }
}

// LWW-Element-Set (支持添加和删除)
class LWWElementSet {
  constructor() {
    this.addSet = new Map();  // 添加集合
    this.removeSet = new Map();  // 删除集合
  }

  add(element, timestamp = Date.now()) {
    this.addSet.set(element, timestamp);
  }

  remove(element, timestamp = Date.now()) {
    this.removeSet.set(element, timestamp);
  }

  get() {
    const elements = new Set();

    for (const [element, addedAt] of this.addSet) {
      const removedAt = this.removeSet.get(element);

      if (!removedAt || addedAt > removedAt) {
        elements.add(element);
      }
    }

    return Array.from(elements);
  }

  merge(other) {
    // 合并添加集合
    for (const [element, timestamp] of other.addSet) {
      const current = this.addSet.get(element);
      if (!current || timestamp > current) {
        this.addSet.set(element, timestamp);
      }
    }

    // 合并删除集合
    for (const [element, timestamp] of other.removeSet) {
      const current = this.removeSet.get(element);
      if (!current || timestamp > current) {
        this.removeSet.set(element, timestamp);
      }
    }
  }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65

// 使用Yjs构建协作应用

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

class CollaborativeDocument {
  constructor() {
    // Y.Doc文档
    this.doc = new Y.Doc();

    // 获取文本类型
    this.text = this.doc.getText('content');

    // WebSocket提供者（实时同步）
    this.wsProvider = new WebsocketProvider(
      'ws://localhost:1234',
      'room-1',
      this.doc
    );

    // IndexedDB持久化（离线存储）
    this.idbProvider = new IndexeddbPersistence(
      'room-1',
      this.doc
    );
  }

  async init() {
    // 等待WebSocket连接
    this.wsProvider.on('sync', (status) => {
      console.log('Sync status:', status);
    });

    // 等待IndexedDB加载
    await this.idbProvider.whenSynced;

    // 监听变化
    this.doc.on('update', (update) => {
      this.handleUpdate(update);
    });
  }

  insert(position, text) {
    this.text.insert(position, text);
  }

  delete(position, length) {
    this.text.delete(position, length);
  }

  get content() {
    return this.text.toString();
  }

  handleUpdate(update) {
    // 处理文档更新
    Y.encodeStateAsUpdate(this.doc);
  }

  // 断开连接
  disconnect() {
    this.wsProvider.disconnect();
  }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43

// WebSocket实时同步

class RealtimeSyncEngine {
  constructor(url) {
    this.ws = new WebSocket(url);
    this.messageQueue = [];
    this.isConnected = false;

    this.ws.onopen = () => {
      this.isConnected = true;
      this.flushMessageQueue();
    };

    this.ws.onmessage = (event) => {
      this.handleMessage(event.data);
    };

    this.ws.onclose = () => {
      this.isConnected = false;
      this.reconnect();
    };
  }

  send(message) {
    if (this.isConnected) {
      this.ws.send(JSON.stringify(message));
    } else {
      this.messageQueue.push(message);
    }
  }

  flushMessageQueue() {
    while (this.messageQueue.length > 0) {
      this.ws.send(JSON.stringify(this.messageQueue.shift()));
    }
  }

  reconnect() {
    setTimeout(() => {
      this.ws = new WebSocket(this.ws.url);
    }, 1000);
  }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41

// 离线优先同步

class OfflineFirstSync {
  constructor() {
    this.localDB = new LocalDatabase();
    this.remoteAPI = new RemoteAPI();
  }

  async write(data) {
    // 1. 写入本地
    const writeTime = Date.now();
    await this.localDB.write({
      data,
      writeTime,
      synced: false
    });

    // 2. 尝试同步到服务器
    try {
      await this.syncToServer(data);
      await this.localDB.markSynced(writeTime);
    } catch (error) {
      // 网络失败，标记为待同步
      console.log('Sync failed, will retry later');
    }
  }

  async syncPendingChanges() {
    const pending = await this.localDB.getPending();

    for (const change of pending) {
      try {
        await this.syncToServer(change.data);
        await this.localDB.markSynced(change.writeTime);
      } catch (error) {
        // 继续处理下一个
        continue;
      }
    }
  }
}