当前位置：首页 > news >正文

CRDT在分布式协作系统中的工程实践：基于RGA算法的实时协同编辑方案

news 2026/7/13 11:26:51

CRDT在分布式协作系统中的工程实践：基于RGA算法的实时协同编辑方案

一、当两个用户同时编辑同一行文本：冲突解决的根本困境

分布式协同编辑系统中，用户A在"hello world"的第6位插入"beautiful "，用户B同时在相同位置删除"world"。两个操作在各自的本地副本上立即执行——等它们通过网络同步到对方时，文本状态已经分叉。

传统的OT（Operational Transformation）方案需要中心服务器进行操作变换，存在单点故障和延迟问题。更致命的是，OT的正确性证明极难验证——已验证的OT算法在N>3节点时可能失去收敛性。CRDT（Conflict-free Replicated Data Type）提供了去中心化的解决方案：通过精确定义的数据结构和合并规则，保证所有副本最终收敛到一致状态。

二、RGA（Replicated Growable Array）算法的核心原理

sequenceDiagram participant A as 用户A (副本1) participant N as 网络 participant B as 用户B (副本2) Note over A,B: 初始状态: "abc" A->>A: insert(pos=2, char='x', id=1) Note over A: "abxc" (本地立即应用) B->>B: insert(pos=1, char='y', id=2) Note over B: "aybc" (本地立即应用) A->>N: 广播 op(id=1) B->>N: 广播 op(id=2) N->>B: op(id=1) B->>B: merge op(id=1) Note over B: 通过id和tombstone计算最终位置 N->>A: op(id=2) A->>A: merge op(id=2) Note over A,B: 最终收敛: "aybxc"

RGA的核心思想：每个字符有全局唯一的ID（由客户端ID+逻辑时钟生成），位置由前驱字符的ID决定而非绝对索引。当两个并发插入操作争抢同一位置时，通过ID的字典序决定先后顺序——ID大的排在后面。删除操作不真正移出字符，而是标记为tombstone。

RGA 的 tombstone 策略在长时间运行的协同编辑系统中会引入一个渐进式性能退化：文档大小的单向增长。每次删除操作仅标记字符为 tombstone，不释放存储空间。对于一个活跃的协作文档，如果用户频繁编辑和删除文本，tombstone 比例可能超过有效字符的 300%。这导致两个实际问题：一是内存占用持续攀升——对于协同代码编辑器，一个 100KB 的源文件可能因多次编辑重写而产生 1MB 的 tombstone 数据；二是全序合并（merge）时需要遍历所有 tombstone，O(n) 的复杂度随文档历史线性增长。解决方案是tombstone 压缩——当文档的 tombstone 比例超过阈值（如 50%）时，触发压缩操作：创建文档快照，释放所有 tombstone 节点的内存，并生成一个新的"压缩锚点"，后续的 ID 基于该锚点重新生成。但压缩破坏了 ID 的历史连续性，需要在所有副本间通过抗熵协议（anti-entropy）同步压缩点——这引入了一致性协议的额外复杂性。Yjs（基于 YATA 算法的 CRDT）通过"墓碑删除+GC 回退"的两阶段方案处理这个问题，RGA 的工程实现同样需要这一层。

三、RGA的Rust工程实现

use std::collections::BTreeMap; use std::fmt; /// 全局唯一标识符：客户端ID + 单调递增计数器 /// 使用(client_id, seq)而非UUID：序列可比较、占用空间更小 #[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)] struct NodeId { client_id: u64, seq: u64, // Lamport时钟值 } /// RGA节点：链表节点，通过前驱ID构建全序关系 #[derive(Debug, Clone)] struct RgaNode { id: NodeId, value: Option<char>, // None表示tombstone（已删除） // 前驱节点ID——这是RGA顺序的核心 // 前驱定义为"在此字符之前逻辑上紧邻的字符" prev_id: Option<NodeId>, } /// RGA文本表示：有序映射——按前驱关系排序 struct RgaText { // 使用BTreeMap按NodeId排序——ID唯一且全序 nodes: BTreeMap<NodeId, RgaNode>, // 每个客户端的逻辑时钟 clocks: Vec<u64>, client_id: u64, } impl RgaText { fn new(client_id: u64, total_clients: usize) -> Self { Self { nodes: BTreeMap::new(), clocks: vec![0; total_clients], client_id, } } /// 生成下一个ID：Lamport时钟递增 fn next_id(&mut self) -> NodeId { let seq = self.clocks[self.client_id as usize]; self.clocks[self.client_id as usize] += 1; NodeId { client_id: self.client_id, seq, } } /// 本地插入：立即生效，生成操作 /// pos是逻辑位置（tombstone不计入），通过迭代器计算 fn local_insert(&mut self, pos: usize, ch: char) -> InsertOp { let new_id = self.next_id(); // 查找插入位置的前驱节点 // 前驱ID决定排序位置，而非整数索引 let (prev_id, _) = self.find_visible_before(pos); let node = RgaNode { id: new_id, value: Some(ch), prev_id, }; self.nodes.insert(new_id, node); InsertOp { char_id: new_id, value: ch, prev_id, lamport_ts: self.clocks[self.client_id as usize], } } /// 查找第pos个可见字符的id和前驱 fn find_visible_before(&self, pos: usize) -> (Option<NodeId>, Option<NodeId>) { let mut prev: Option<NodeId> = None; let mut visible_count = 0; // 按ID顺序遍历：保证确定性排序 for (id, node) in &self.nodes { if node.value.is_some() { if visible_count == pos { return (prev, Some(*id)); } prev = Some(*id); visible_count += 1; } } // 插入到末尾 (prev, None) } /// 合并远程操作：确定性规则保证最终一致性 /// 两个并发插入通过id比较决定顺序 fn merge(&mut self, op: InsertOp) -> MergeResult { // 检查是否已应用（幂等性） if self.nodes.contains_key(&op.char_id) { return MergeResult::AlreadyApplied; } // 更新逻辑时钟 self.clocks[op.char_id.client_id as usize] = self.clocks [op.char_id.client_id as usize] .max(op.lamport_ts); // 查找正确的前驱位置 let effective_prev = self.resolve_effective_prev(&op); let node = RgaNode { id: op.char_id, value: Some(op.value), prev_id: effective_prev, }; self.nodes.insert(op.char_id, node); MergeResult::Applied } /// 解析有效前驱：处理并发情况 /// 如果前驱已被删除或被其他插入"挤开"，需要重新计算 fn resolve_effective_prev(&self, op: &InsertOp) -> Option<NodeId> { // 前驱可能被标记为tombstone——仍然有效（位置引用） // 也可能有其他并发插入占据同一前驱位置 if let Some(prev) = op.prev_id { if self.nodes.contains_key(&prev) { return Some(prev); } } // 前驱不存在：找到id小于op.char_id的最大可见节点 self.nodes .iter() .rev() .find(|(id, node)| { node.value.is_some() && *id < &op.char_id }) .map(|(id, _)| *id) } /// 本地删除：标记tombstone，不物理移除 fn local_delete(&mut self, pos: usize) -> Option<DeleteOp> { let (_prev, target) = self.find_visible_before(pos); let target = target?; if let Some(node) = self.nodes.get_mut(&target) { if node.value.is_some() { node.value = None; return Some(DeleteOp { char_id: target }); } } None } /// 获取可见文本：跳过tombstone fn to_string(&self) -> String { self.nodes .iter() .filter_map(|(_, node)| node.value) .collect() } /// 获取文本长度：不含tombstone fn visible_len(&self) -> usize { self.nodes.values().filter(|n| n.value.is_some()).count() } } /// 插入操作：通过网络同步的数据结构 #[derive(Debug, Clone)] struct InsertOp { char_id: NodeId, value: char, prev_id: Option<NodeId>, lamport_ts: u64, } /// 删除操作 #[derive(Debug, Clone)] struct DeleteOp { char_id: NodeId, } /// 合并结果枚举 enum MergeResult { Applied, AlreadyApplied, } /// 协同编辑会话管理 struct CollaborativeSession { document: RgaText, // 待发送的操作队列 pending_ops: Vec<InsertOp>, } impl CollaborativeSession { /// Tombstone清理：定期压缩历史 /// 保留期后（如24小时）的tombstone可安全移除 /// 前提：所有客户端已确认在保留期前的操作 fn compact_tombstones(&mut self, _retention_period_ms: u64) -> usize { let before = self.document.nodes.len(); // 简化实现：移除tombstone self.document.nodes.retain(|_, node| { node.value.is_some() }); before - self.document.nodes.len() } } // 确定性排序演示 impl Ord for RgaText { fn cmp(&self, other: &Self) -> std::cmp::Ordering { // 两个RGA副本通过to_string比较是否收敛 self.to_string().cmp(&other.to_string()) } } impl PartialOrd for RgaText { fn partial_cmp(&self, other: &Self) -> Option<std::cmp::Ordering> { Some(self.cmp(other)) } } impl PartialEq for RgaText { fn eq(&self, other: &Self) -> bool { self.to_string() == other.to_string() } } impl Eq for RgaText {}

设计决策解析：