JS `CRDT` (Conflict-Free Replicated Data Types) 算法在实时协作中的实现细节

大家好，欢迎来到今天的“CRDT：实时协作的魔法棒”讲座！今天咱们不讲虚的，直接撸起袖子，用代码和人话，把CRDT这玩意儿扒个底朝天，看看它到底是怎么在实时协作里呼风唤雨的。

开场白：实时协作，痛点在哪里？

想象一下，你和你的小伙伴正在愉快地在线编辑同一份文档。你敲了一段话，他删了一段字，如果服务器简单粗暴地按照接收到的顺序应用这些操作，那画面太美我不敢看。轻则文档错乱，重则引发世界大战（夸张手法）。

所以，实时协作的关键在于：如何保证在网络延迟、离线操作等情况下，不同客户端最终都能达成一致的状态？

传统的做法，比如Operational Transformation (OT)，虽然能解决部分问题，但复杂度高，调试困难，而且容易出现各种边缘情况。而CRDT，则提供了一种更优雅、更可靠的解决方案。

CRDT：闪亮登场！

CRDT，全称Conflict-Free Replicated Data Type，中文名叫“无冲突复制数据类型”。听起来高大上，其实核心思想很简单：让数据自己解决冲突，而不是依赖服务器。

CRDT分两种主要类型：

1. State-based CRDT (CvRDT)： 基于状态的CRDT。客户端直接交换完整的数据状态，然后通过一个合并函数来解决冲突。
2. Operation-based CRDT (CmRDT)： 基于操作的CRDT。客户端广播操作，然后每个客户端按照相同的顺序应用这些操作。

CvRDT：状态同步，简单粗暴但有效

CvRDT的核心在于交换状态和合并状态。每个客户端都维护一个数据的完整状态，当需要同步时，就将自己的状态发送给其他客户端。其他客户端收到状态后，使用一个合并函数将收到的状态和自己的状态合并，得到一个新的状态。

听起来有点抽象？咱们来个例子：

例1: Grow-Only Counter (只增计数器)

这是最简单的CvRDT之一。每个客户端维护一个计数器，只能增加，不能减少。

class GrowOnlyCounter {
  constructor(replicaId) {
    this.replicaId = replicaId; // 每个副本的唯一ID
    this.count = 0;
  }

  increment(amount) {
    this.count += amount;
  }

  getState() {
    return {
      replicaId: this.replicaId,
      count: this.count,
    };
  }

  merge(otherCounter) {
    this.count = Math.max(this.count, otherCounter.count);
  }
}

// 例子
const counter1 = new GrowOnlyCounter("A");
const counter2 = new GrowOnlyCounter("B");

counter1.increment(5);
counter2.increment(10);

// 同步：counter1 接收 counter2 的状态
counter1.merge(counter2.getState());
console.log("Counter1 count after merge:", counter1.count); // 输出：10

// 反之，counter2 接收 counter1 的状态
counter2.merge(counter1.getState());
console.log("Counter2 count after merge:", counter2.count); // 输出：10

代码解释：

• replicaId：每个计数器都有一个唯一的ID，用来标识它是哪个客户端的。
• increment()：增加计数器的值。
• getState()：返回当前计数器的状态。
• merge()：合并两个计数器的状态，取最大值。

重点： merge()函数是CvRDT的核心。它必须满足交换律、结合律和幂等律。

• 交换律： merge(a, b) 等于 merge(b, a)
• 结合律： merge(a, merge(b, c)) 等于 merge(merge(a, b), c)
• 幂等律： merge(a, a) 等于 a

满足这些定律，才能保证无论状态合并的顺序如何，最终的结果都是一致的。

例2: Last Write Wins (LWW) Element Set (后写胜出集合)

LWW Element Set 是一种常见的集合类型，允许添加和删除元素。每个元素都有一个时间戳，当添加和删除操作冲突时，时间戳更大的操作胜出。

class LWWElementSet {
  constructor() {
    this.adds = new Map(); // 存储添加的元素，key是元素，value是时间戳
    this.removes = new Map(); // 存储删除的元素，key是元素，value是时间戳
  }

  add(element, timestamp) {
    this.adds.set(element, timestamp);
  }

  remove(element, timestamp) {
    this.removes.set(element, timestamp);
  }

  lookup(element) {
    const addTimestamp = this.adds.get(element) || 0;
    const removeTimestamp = this.removes.get(element) || 0;
    return addTimestamp > removeTimestamp; // 如果添加时间戳大于删除时间戳，则元素存在
  }

  getState() {
    return {
      adds: new Map(this.adds), // 深拷贝，避免直接修改内部状态
      removes: new Map(this.removes),
    };
  }

  merge(otherSet) {
    // 合并 adds
    for (const [element, timestamp] of otherSet.adds) {
      const currentTimestamp = this.adds.get(element) || 0;
      if (timestamp > currentTimestamp) {
        this.adds.set(element, timestamp);
      }
    }

    // 合并 removes
    for (const [element, timestamp] of otherSet.removes) {
      const currentTimestamp = this.removes.get(element) || 0;
      if (timestamp > currentTimestamp) {
        this.removes.set(element, timestamp);
      }
    }
  }
}

// 例子
const set1 = new LWWElementSet();
const set2 = new LWWElementSet();

set1.add("apple", 1);
set2.add("apple", 2); // 后添加的 "apple" 胜出
set1.remove("apple", 3); // 后删除的 "apple" 胜出

set1.merge(set2.getState());
console.log("Set1 contains apple:", set1.lookup("apple")); // 输出：false

set2.merge(set1.getState());
console.log("Set2 contains apple:", set2.lookup("apple")); // 输出：false

代码解释：

• adds 和 removes 分别存储添加和删除的元素，以及对应的时间戳。
• lookup()：判断元素是否存在，比较添加和删除的时间戳。
• merge()：合并两个集合，保留时间戳更大的添加和删除操作。

CvRDT的优缺点：

• 优点： 简单易懂，易于实现。
• 缺点： 需要传输完整状态，数据量大，效率较低。不适合大型数据。

特性	CvRDT
数据传输量	大，传输完整状态
复杂度	相对较低
适用场景	数据量小，状态更新频率不高

CmRDT：操作广播，精准高效

CmRDT的核心在于广播操作和保证操作顺序。每个客户端将自己的操作广播给其他客户端，然后每个客户端按照相同的顺序应用这些操作。

例3: Add-wins Last Write Wins Element Set (Add-wins LWW Element Set)

与LWW Element Set类似，但Add操作总是胜出，即使Remove操作的时间戳更大。

class AddWinsLWWElementSet {
  constructor() {
    this.adds = new Map();
    this.removes = new Map();
  }

  add(element, timestamp) {
    this.adds.set(element, timestamp);
  }

  remove(element, timestamp) {
    this.removes.set(element, timestamp);
  }

  lookup(element) {
    const addTimestamp = this.adds.get(element) || 0;
    const removeTimestamp = this.removes.get(element) || 0;
    return addTimestamp >= removeTimestamp; // Add操作总是胜出
  }

  // CmRDT 需要序列化操作
  serializeAdd(element, timestamp) {
    return { type: 'add', element, timestamp };
  }

  serializeRemove(element, timestamp) {
    return { type: 'remove', element, timestamp };
  }

  applyOperation(operation) {
    switch (operation.type) {
      case 'add':
        this.add(operation.element, operation.timestamp);
        break;
      case 'remove':
        this.remove(operation.element, operation.timestamp);
        break;
    }
  }

  getState() {
      // 注意: CmRDT 一般不需要 getState 和 merge 方法，这里是为了演示完整性
      return {
        adds: new Map(this.adds),
        removes: new Map(this.removes),
      };
    }

    merge(otherSet) {
      // 注意: CmRDT 一般不需要 getState 和 merge 方法，这里是为了演示完整性
      for (const [element, timestamp] of otherSet.adds) {
        const currentTimestamp = this.adds.get(element) || 0;
        if (timestamp > currentTimestamp) {
          this.adds.set(element, timestamp);
        }
      }

      for (const [element, timestamp] of otherSet.removes) {
        const currentTimestamp = this.removes.get(element) || 0;
        if (timestamp > currentTimestamp) {
          this.removes.set(element, timestamp);
        }
      }
    }
}

// 例子
const set1 = new AddWinsLWWElementSet();
const set2 = new AddWinsLWWElementSet();

// 模拟操作广播
const addOperation = set1.serializeAdd("apple", 1);
const removeOperation = set2.serializeRemove("apple", 2);

// 应用操作 (保证顺序)
set1.applyOperation(addOperation);
set1.applyOperation(removeOperation);
console.log("Set1 contains apple:", set1.lookup("apple")); // 输出：true (Add wins)

set2.applyOperation(addOperation);
set2.applyOperation(removeOperation);
console.log("Set2 contains apple:", set2.lookup("apple")); // 输出：true (Add wins)

代码解释：

• serializeAdd() 和 serializeRemove()：将操作序列化，方便广播。
• applyOperation()：应用操作，根据操作类型执行相应的添加或删除。
• Add Wins 的核心在于 lookup 函数，添加操作总是胜出。

重点： CmRDT需要保证操作的顺序。通常可以使用向量时钟或序列号来保证操作的顺序。

向量时钟： 每个客户端维护一个向量，记录每个客户端的操作数量。当客户端发送操作时，会带上当前的向量。接收方根据向量来确定操作的顺序。

序列号： 每个客户端维护一个序列号，每次发送操作时，序列号递增。接收方根据序列号来确定操作的顺序。

CmRDT的优缺点：

• 优点： 数据传输量小，效率高。
• 缺点： 实现复杂，需要保证操作的顺序。

特性	CmRDT
数据传输量	小，只传输操作
复杂度	相对较高，需要保证操作顺序
适用场景	数据量大，状态更新频繁

选择哪种CRDT？

选择哪种CRDT取决于具体的应用场景。

• 如果数据量小，状态更新频率不高，可以选择CvRDT。
• 如果数据量大，状态更新频繁，可以选择CmRDT。

实际应用：文本编辑器

CRDT在文本编辑器中的应用比较复杂，但核心思想仍然是让数据自己解决冲突。

一种常见的做法是使用Yjs库。Yjs是一个基于CRDT的协作框架，提供了多种CRDT类型，可以用来实现文本编辑器的各种功能，比如插入、删除、替换等。

Yjs使用一种叫做Relative Positioning的技术来解决文本插入位置的冲突。每个插入位置都相对于文档中的其他位置，而不是绝对的位置。这样，即使文档中的其他位置发生了变化，插入位置仍然有效。

总结：CRDT的优势

CRDT的优势在于：

• 最终一致性： 保证所有客户端最终都能达成一致的状态。
• 无需中心协调： 不需要中心服务器来协调冲突，降低了系统的复杂度。
• 离线操作： 允许客户端在离线状态下进行操作，并在上线后自动同步。
• 容错性： 即使部分客户端发生故障，系统仍然可以正常工作。

表格：CvRDT vs CmRDT

特性	CvRDT	CmRDT
数据传输量	大，传输完整状态	小，只传输操作
复杂度	相对较低	相对较高，需要保证操作顺序
状态同步方式	交换完整状态，然后合并	广播操作，每个客户端按相同顺序应用
适用场景	数据量小，状态更新频率不高	数据量大，状态更新频繁
例子	Grow-Only Counter, LWW Element Set	Add-wins LWW Element Set
保证一致性方式	合并函数满足交换律、结合律、幂等律	向量时钟、序列号，保证操作顺序

结束语：CRDT，实时协作的未来

CRDT是一种强大的工具，可以用来解决实时协作中的各种难题。虽然CRDT的实现比较复杂，但它的优势是显而易见的。随着实时协作应用的普及，CRDT将会扮演越来越重要的角色。

希望今天的讲座对大家有所帮助！ 记住，代码才是王道，多动手实践才能真正理解CRDT的精髓。 下课！ (手动狗头)