上一讲我们构建了文档引擎的核心——Piece Table 文本模型、操作历史、光标管理。但这些都是单用户视角。真正的协同编辑需要解决一个根本问题两个人同时编辑同一位置如何保证最终结果一致这就是 Operational TransformationOT 算法要解决的问题。一、OT 算法概述1.1 问题场景用户A和用户B同时编辑文档 Hello 用户A: 在位置5插入 World → 期望: Hello World 用户B: 在位置5插入 !!! → 期望: Hello !!! 如果简单按顺序应用: A先执行: Hello → Hello World B再执行: 在位置5插入 !!! → Hello!!! World ❌ 错误! 正确结果应该是: Hello World!!! 或 Hello !!! World1.2 OT 的核心思想OT 算法的本质 当两个操作并发发生时不直接应用对方的操作 而是先将对方的操作变换Transform到自己的上下文上 然后再应用。 操作变换公式 transform(op_a, op_b) → (op_a, op_b) 其中 op_a 是在 op_b 之后执行时等价于 op_a 的操作 op_b 是在 op_a 之后执行时等价于 op_b 的操作 性质无论谁先执行最终结果一致1.3 OT 的类型类型说明代表系统COT​客户端-服务端 OTGoogle WaveJupiter​双向同步 OTGoogle Docs 早期GOTO​分组 OT通用模型Tombstone​墓碑变换ShareJS我们采用Jupiter 模型它简洁且适合 WebSocket 双向通信。二、操作定义与代数2.1 操作的形式化# core/ot/operation.py OT 操作的形式化定义 from __future__ import annotations from dataclasses import dataclass, field from enum import Enum from typing import Optional, Tuple import time import uuid class OpType(Enum): 操作类型 INSERT insert DELETE delete NOOP noop # 空操作 dataclass class OTOperation: OT 操作 比基础 Operation 多了 site_id 和 sequence 信息 用于分布式环境下的操作排序和追踪。 # 操作信息 op_type: OpType position: int text: str deleted_text: str # 分布式标识 site_id: str # 产生此操作的站点 sequence: int 0 # 站点内的操作序号 timestamp: float 0.0 # 操作时间戳 # 操作依赖 depends_on: Optional[str] None # 依赖的上一个操作 ID def __post_init__(self): if not self.timestamp: self.timestamp time.time() if not self.site_id: self.site_id fsite-{uuid.uuid4().hex[:8]} property def id(self) - str: 操作唯一标识 return f{self.site_id}:{self.sequence} def clone(self) - OTOperation: 克隆操作 return OTOperation( op_typeself.op_type, positionself.position, textself.text, deleted_textself.deleted_text, site_idself.site_id, sequenceself.sequence, timestampself.timestamp ) def shift_position(self, delta: int): 偏移操作位置用于变换 self.position delta if self.position 0: self.position 0 def to_dict(self) - dict: 序列化 return { op_type: self.op_type.value, position: self.position, text: self.text, deleted_text: self.deleted_text, site_id: self.site_id, sequence: self.sequence, timestamp: self.timestamp, id: self.id } classmethod def from_dict(cls, data: dict) - OTOperation: 反序列化 return cls( op_typeOpType(data[op_type]), positiondata[position], textdata.get(text, ), deleted_textdata.get(deleted_text, ), site_iddata.get(site_id, ), sequencedata.get(sequence, 0), timestampdata.get(timestamp, 0.0) ) def __repr__(self) - str: if self.op_type OpType.INSERT: return fINS({self.text} {self.position})[{self.id}] elif self.op_type OpType.DELETE: return fDEL(-{self.deleted_text} {self.position})[{self.id}] return fNOOP[{self.id}]三、变换函数实现3.1 核心变换逻辑# core/ot/transform.py OT 变换函数 实现操作之间的变换保证并发操作的一致性。 from .operation import OTOperation, OpType import logging logger logging.getLogger(__name__) def transform(op1: OTOperation, op2: OTOperation) - tuple: 操作变换函数 将 op1 和 op2 变换为可以在对方之后执行的形式。 Args: op1: 第一个操作假设已执行 op2: 第二个操作待变换 Returns: (op1, op2) 变换后的操作对 其中 op1 等价于在 op2 之后执行的 op1 op2 等价于在 op1 之后执行的 op2 # 如果是 NOOP直接返回 if op1.op_type OpType.NOOP: return op1.clone(), op2.clone() if op2.op_type OpType.NOOP: return op1.clone(), op2.clone() # 根据操作组合选择变换规则 if op1.op_type OpType.INSERT and op2.op_type OpType.INSERT: return _transform_insert_insert(op1, op2) elif op1.op_type OpType.INSERT and op2.op_type OpType.DELETE: return _transform_insert_delete(op1, op2) elif op1.op_type OpType.DELETE and op2.op_type OpType.INSERT: return _transform_delete_insert(op1, op2) elif op1.op_type OpType.DELETE and op2.op_type OpType.DELETE: return _transform_delete_delete(op1, op2) return op1.clone(), op2.clone() def _transform_insert_insert(op1: OTOperation, op2: OTOperation) - tuple: INSERT vs INSERT 变换 两人在同一位置插入不同文本 - 如果 op1 的位置 op2 的位置op2 的位置后移 op1 的长度 - 如果 op1 的位置 op2 的位置op1 的位置后移 op2 的长度 - 如果位置相同约定 site_id 小的先执行 op1_prime op1.clone() op2_prime op2.clone() if op1.position op2.position: # op1 在 op2 前面插入op2 的位置需要后移 op2_prime.shift_position(len(op1.text)) elif op1.position op2.position: # op2 在 op1 前面插入op1 的位置需要后移 op1_prime.shift_position(len(op2.text)) else: # 同一位置按 site_id 排序 if op1.site_id op2.site_id: op2_prime.shift_position(len(op1.text)) else: op1_prime.shift_position(len(op2.text)) logger.debug(fXFORM I/I: {op1} {op2} → ({op1_prime}, {op2_prime})) return op1_prime, op2_prime def _transform_insert_delete(op1: OTOperation, op2: OTOperation) - tuple: INSERT vs DELETE 变换 插入操作遇到删除操作 - 如果插入位置在删除范围内插入位置调整到删除起始位置 - 如果插入位置在删除范围之后插入位置前移删除长度 op1_prime op1.clone() op2_prime op2.clone() delete_start op2.position delete_end op2.position len(op2.deleted_text) if op1.position delete_start: # 插入在删除之前不受影响 pass elif op1.position delete_end: # 插入在删除范围内 → 移动到删除起始位置 op1_prime.position delete_start else: # 插入在删除之后 → 位置前移 op1_prime.shift_position(-len(op2.deleted_text)) logger.debug(fXFORM I/D: {op1} {op2} → ({op1_prime}, {op2_prime})) return op1_prime, op2_prime def _transform_delete_insert(op1: OTOperation, op2: OTOperation) - tuple: DELETE vs INSERT 变换 删除操作遇到插入操作 - 如果删除范围在插入之后删除位置后移插入长度 - 如果删除范围包含插入位置删除范围扩大 op1_prime op1.clone() op2_prime op2.clone() delete_start op1.position delete_end op1.position len(op1.deleted_text) insert_len len(op2.text) if op2.position delete_start: # 插入在删除之前 → 删除位置后移 op1_prime.shift_position(insert_len) elif op2.position delete_end: # 插入在删除范围内 → 删除范围扩大 op1_prime.deleted_text ( op1.deleted_text[:op2.position - delete_start] op2.text op1.deleted_text[op2.position - delete_start:] ) else: # 插入在删除之后不受影响 pass logger.debug(fXFORM D/I: {op1} {op2} → ({op1_prime}, {op2_prime})) return op1_prime, op2_prime def _transform_delete_delete(op1: OTOperation, op2: OTOperation) - tuple: DELETE vs DELETE 变换 两人删除同一区域 - 重叠部分只删除一次 - 非重叠部分各自调整位置 op1_prime op1.clone() op2_prime op2.clone() d1_start op1.position d1_end op1.position len(op1.deleted_text) d2_start op2.position d2_end op2.position len(op2.deleted_text) # 情况 1: op1 完全在 op2 之前 if d1_end d2_start: op2_prime.shift_position(-len(op1.deleted_text)) # 情况 2: op1 完全在 op2 之后 elif d1_start d2_end: op1_prime.shift_position(-len(op2.deleted_text)) # 情况 3: 有重叠 else: overlap_start max(d1_start, d2_start) overlap_end min(d1_end, d2_end) overlap_len overlap_end - overlap_start # 调整 op1 的删除范围 if d1_start d2_start: # op1 延伸到 op2 的起始 op1_prime.deleted_text op1.deleted_text[:d2_start - d1_start] elif d1_start d2_start: # op1 从 op2 内部开始 op1_prime.position d2_start op1_prime.deleted_text op1.deleted_text[d1_start - d2_start:] # 调整 op2 的删除范围 if d2_start d1_start: op2_prime.deleted_text op2.deleted_text[:d1_start - d2_start] elif d2_start d1_start: op2_prime.position d1_start op2_prime.deleted_text op2.deleted_text[d2_start - d1_start:] logger.debug(fXFORM D/D: {op1} {op2} → ({op1_prime}, {op2_prime})) return op1_prime, op2_prime def compose(op1: OTOperation, op2: OTOperation) - OTOperation: 操作组合 将两个连续的操作合并为一个等价操作。 用于批量同步时减少操作数量。 if op1.op_type OpType.NOOP: return op2.clone() if op2.op_type OpType.NOOP: return op1.clone() # INSERT INSERT if op1.op_type OpType.INSERT and op2.op_type OpType.INSERT: if op2.position op1.position len(op1.text): # op2 在 op1 后面 result OTOperation( OpType.INSERT, op1.position, textop1.text op2.text, site_idop1.site_id, sequenceop1.sequence ) return result # DELETE DELETE if op1.op_type OpType.DELETE and op2.op_type OpType.DELETE: if op2.position op1.position: # 连续删除 result OTOperation( OpType.DELETE, op1.position, deleted_textop1.deleted_text op2.deleted_text, site_idop1.site_id, sequenceop1.sequence ) return result # 无法组合返回 op2 return op2.clone() def invert(op: OTOperation) - OTOperation: 操作求逆 生成操作的逆操作用于撤销。 if op.op_type OpType.INSERT: return OTOperation( OpType.DELETE, op.position, deleted_textop.text, site_idop.site_id, sequenceop.sequence ) elif op.op_type OpType.DELETE: return OTOperation( OpType.INSERT, op.position, textop.deleted_text, site_idop.site_id, sequenceop.sequence ) return OTOperation(OpType.NOOP, 0)四、Jupiter 同步模型4.1 模型架构Jupiter 同步模型 客户端 服务端 │ │ │─── op_A (seq1) ──────▶│ (发送操作) │ │ │◀── ack (seq1) ───────│ (确认) │ │ │─── op_B (seq2) ──────▶│ │ │ │◀── op_C (from other) ──│ (接收远程操作) │ │ │─── transform(op_C) ────│ (变换后应用) │ │ 关键数据结构 - 客户端维护一个未确认操作列表pending - 服务端维护每个客户端的已收到操作列表 - 新操作到达时与列表中所有操作进行变换4.2 客户端同步引擎# core/ot/sync_engine.py Jupiter 同步引擎 实现客户端和服务端的操作同步逻辑。 from __future__ import annotations from typing import List, Optional, Callable from .operation import OTOperation, OpType from .transform import transform import logging import threading import time logger logging.getLogger(__name__) class SyncEngine: 同步引擎Jupiter 模型 负责 1. 管理未确认操作列表 2. 执行操作变换 3. 维护操作顺序 def __init__(self, site_id: str, on_apply: Optional[Callable] None): Args: site_id: 站点标识 on_apply: 应用操作的回调函数 self.site_id site_id self.on_apply on_apply # 未确认操作列表等待服务端确认 self.pending: List[OTOperation] [] # 已确认操作计数 self.confirmed_ops 0 # 本地操作序号 self.local_seq 0 # 锁 self.lock threading.Lock() # ---------- 本地操作 ---------- def generate_local_op(self, op_type: OpType, position: int, text: str , deleted_text: str ) - OTOperation: 生成本地操作 Args: op_type: 操作类型 position: 位置 text: 插入文本 deleted_text: 删除文本 Returns: 生成的 OT 操作 with self.lock: self.local_seq 1 op OTOperation( op_typeop_type, positionposition, texttext, deleted_textdeleted_text, site_idself.site_id, sequenceself.local_seq ) # 加入未确认列表 self.pending.append(op) logger.debug(fLocal op generated: {op}) return op def apply_local_op(self, op: OTOperation): 应用本地操作立即生效 if self.on_apply: self.on_apply(op) # ---------- 远程操作处理 ---------- def receive_remote_op(self, remote_op: OTOperation) - Optional[OTOperation]: 接收远程操作 将远程操作与未确认操作进行变换得到可以在本地执行的形式。 Args: remote_op: 远程操作 Returns: 变换后的操作可在本地执行 with self.lock: transformed_op remote_op.clone() # 与所有未确认操作进行变换 for pending_op in self.pending: # 变换: (pending_op, transformed_op) → (_, transformed_op) _, transformed_op transform(pending_op, transformed_op) logger.debug(fRemote op transformed: {remote_op} → {transformed_op}) return transformed_op def acknowledge_op(self, seq: int): 确认操作 服务端确认收到某个操作后从未确认列表中移除。 with self.lock: # 找到并移除已确认的操作 for i, op in enumerate(self.pending): if op.sequence seq: self.pending self.pending[i 1:] self.confirmed_ops 1 logger.debug(fOp acknowledged: {op}) break def get_pending_count(self) - int: 获取未确认操作数 with self.lock: return len(self.pending) def get_status(self) - dict: 获取同步状态 with self.lock: return { site_id: self.site_id, local_seq: self.local_seq, confirmed_ops: self.confirmed_ops, pending_count: len(self.pending) } class ServerSyncEngine: 服务端同步引擎 管理所有客户端的操作状态执行全局变换。 def __init__(self, document_id: str): self.document_id document_id # 所有已确认的操作按顺序 self.history: List[OTOperation] [] # 每个客户端的状态 self.client_states: dict {} # site_id - ClientState # 锁 self.lock threading.Lock() def register_client(self, site_id: str) - ClientState: 注册客户端 state ClientState(site_id) self.client_states[site_id] state return state def receive_op(self, site_id: str, op: OTOperation) - List[OTOperation]: 接收客户端操作 1. 与服务端历史进行变换 2. 广播给其他客户端 Args: site_id: 发送操作的客户端 op: 操作 Returns: 需要广播给其他客户端的操作列表 with self.lock: client_state self.client_states.get(site_id) if not client_state: logger.warning(fUnknown client: {site_id}) return [] # 与服务端历史进行变换 transformed_op op.clone() for hist_op in self.history: _, transformed_op transform(hist_op, transformed_op) # 添加到历史 self.history.append(transformed_op) # 更新客户端状态 client_state.last_seq op.sequence # 生成需要广播给其他客户端的操作 broadcast_ops [] for other_site, other_state in self.client_states.items(): if other_site ! site_id: # 为其他客户端变换操作 other_op transformed_op.clone() for hist_op in self.history[:-1]: # 排除自身 _, other_op transform(hist_op, other_op) broadcast_ops.append(other_op) logger.debug(fServer received {op}, broadcasting {len(broadcast_ops)} ops) return broadcast_ops def get_history_since(self, site_id: str, since_seq: int) - List[OTOperation]: 获取某个客户端缺失的操作 用于客户端重连后同步。 with self.lock: # 找到该客户端最后一个确认的操作 client_state self.client_states.get(site_id) if not client_state: return list(self.history) # 返回历史中该客户端之后的操作 missing_ops [] for op in self.history: if op.site_id ! site_id or op.sequence client_state.last_seq: missing_ops.append(op) return missing_ops dataclass class ClientState: 客户端状态 site_id: str last_seq: int 0 connected_at: float 0.0 def __post_init__(self): if not self.connected_at: self.connected_at time.time()五、OT 集成到文档引擎5.1 OTDocument 类# core/ot/ot_document.py OT 增强的文档引擎 将 OT 同步集成到 Document 中。 from typing import List, Optional, Callable from ..document import Document from .operation import OTOperation, OpType from .sync_engine import SyncEngine from .transform import transform import logging logger logging.getLogger(__name__) class OTDocument(Document): OT 文档引擎 继承自 Document增加了 OT 同步能力。 def __init__(self, document_id: str, site_id: str, initial_text: str ): super().__init__(document_id, initial_text) self.site_id site_id self.sync_engine SyncEngine( site_idsite_id, on_applyself._apply_op ) # 远程操作回调 self.on_remote_op: Optional[Callable] None # ---------- 本地操作OT 增强 ---------- def insert(self, position: int, text: str) - OTOperation: 插入文本生成 OT 操作 # 1. 生成 OT 操作 ot_op self.sync_engine.generate_local_op( OpType.INSERT, position, texttext ) # 2. 应用本地操作 self.sync_engine.apply_local_op(ot_op) # 3. 更新文档内容 super().insert(position, text) return ot_op def delete(self, position: int, length: int) - OTOperation: 删除文本生成 OT 操作 # 获取被删除的文本 deleted_text self.get_text_range(position, position length) # 1. 生成 OT 操作 ot_op self.sync_engine.generate_local_op( OpType.DELETE, position, deleted_textdeleted_text ) # 2. 应用本地操作 self.sync_engine.apply_local_op(ot_op) # 3. 更新文档内容 super().delete(position, length) return ot_op # ---------- 远程操作 ---------- def receive_remote_op(self, remote_op: OTOperation): 接收并应用远程操作 Args: remote_op: 来自其他站点的操作 # 1. 变换远程操作 transformed_op self.sync_engine.receive_remote_op(remote_op) # 2. 应用到文档 self._apply_op(transformed_op) # 3. 通知回调 if self.on_remote_op: self.on_remote_op(transformed_op) def acknowledge_op(self, seq: int): 确认操作 self.sync_engine.acknowledge_op(seq) def _apply_op(self, op: OTOperation): 应用操作到文档不生成新的 OT 操作 if op.op_type OpType.INSERT: super().insert(op.position, op.text) elif op.op_type OpType.DELETE: super().delete(op.position, len(op.deleted_text)) def get_sync_status(self) - dict: 获取同步状态 return self.sync_engine.get_status()六、测试6.1 OT 变换测试# tests/test_ot.py import pytest from core.ot.operation import OTOperation, OpType from core.ot.transform import ( transform, compose, invert, _transform_insert_insert, _transform_insert_delete, _transform_delete_insert, _transform_delete_delete ) class TestOTTransform: OT 变换函数测试 def test_insert_insert_same_position(self): 同一位置插入 op1 OTOperation(OpType.INSERT, 5, textAB, site_idA) op2 OTOperation(OpType.INSERT, 5, textCD, site_idB) op1_prime, op2_prime transform(op1, op2) # A 先执行B 后移 assert op1_prime.position 5 assert op2_prime.position 7 # 5 2 def test_insert_insert_different_position(self): 不同位置插入 op1 OTOperation(OpType.INSERT, 3, textXX, site_idA) op2 OTOperation(OpType.INSERT, 7, textYY, site_idB) op1_prime, op2_prime transform(op1, op2) # op1 在 op2 前面op2 后移 assert op1_prime.position 3 assert op2_prime.position 9 # 7 2 def test_insert_vs_delete_before(self): 插入在删除之前 op1 OTOperation(OpType.INSERT, 3, textXX) op2 OTOperation(OpType.DELETE, 5, deleted_textHello) op1_prime, op2_prime transform(op1, op2) # 插入在删除之前不受影响 assert op1_prime.position 3 assert op2_prime.position 5 def test_insert_vs_delete_inside(self): 插入在删除范围内 op1 OTOperation(OpType.INSERT, 6, textXX) op2 OTOperation(OpType.DELETE, 5, deleted_textHello) op1_prime, op2_prime transform(op1, op2) # 插入在删除范围内移到删除起始 assert op1_prime.position 5 def test_delete_vs_delete_overlap(self): 重叠删除 op1 OTOperation(OpType.DELETE, 3, deleted_textABCDEF) op2 OTOperation(OpType.DELETE, 5, deleted_textDEFGHI) op1_prime, op2_prime transform(op1, op2) # 重叠部分只删除一次 assert len(op1_prime.deleted_text) len(op1.deleted_text) class TestOTCompose: 操作组合测试 def test_compose_inserts(self): 组合连续插入 op1 OTOperation(OpType.INSERT, 0, textHello) op2 OTOperation(OpType.INSERT, 5, text World) result compose(op1, op2) assert result.text Hello World assert result.position 0 class TestOTInvert: 操作求逆测试 def test_invert_insert(self): 插入的逆操作是删除 op OTOperation(OpType.INSERT, 5, textHello) inv invert(op) assert inv.op_type OpType.DELETE assert inv.position 5 assert inv.deleted_text Hello def test_invert_delete(self): 删除的逆操作是插入 op OTOperation(OpType.DELETE, 3, deleted_textWorld) inv invert(op) assert inv.op_type OpType.INSERT assert inv.position 3 assert inv.text World class TestOTDocument: OT 文档测试 def test_two_users_sync(self): 双用户同步测试 from core.ot.ot_document import OTDocument # 用户 A 和 B 共享同一个文档 doc_a OTDocument(doc-1, site-A, Hello) doc_b OTDocument(doc-1, site-B, Hello) # 用户 A 插入 op_a doc_a.insert(5, World) # 用户 B 插入并发 op_b doc_b.insert(5, !!!) # 互相接收对方操作 doc_a.receive_remote_op(op_b) doc_b.receive_remote_op(op_a) # 最终结果应该一致 assert doc_a.get_text() doc_b.get_text() print(fDoc A: {doc_a.get_text()}) print(fDoc B: {doc_b.get_text()}) if __name__ __main__: pytest.main([__file__, -v])七、可视化演示7.1 OT 工作流演示# examples/ot_demo.py OT 算法工作流演示 from core.ot.operation import OTOperation, OpType from core.ot.transform import transform from core.ot.sync_engine import SyncEngine, ServerSyncEngine from core.ot.ot_document import OTDocument def demonstrate_transform_rules(): 演示变换规则 print( * 65) print( OT 变换规则演示) print( * 65) # 场景 1: 同一位置插入 print(\n 场景 1: 同一位置插入) print(- * 40) doc Hello op_a OTOperation(OpType.INSERT, 5, text World, site_idA, sequence1) op_b OTOperation(OpType.INSERT, 5, text!!!, site_idB, sequence1) print(f初始文档: {doc}) print(f用户 A: {op_a}) print(f用户 B: {op_b}) op_a_prime, op_b_prime transform(op_a, op_b) print(f\n变换后:) print(f A {op_a_prime}) print(f B {op_b_prime}) # 模拟执行 result_a Hello World !!! # A 先插B 后移 result_b Hello !!! World # B 先插A 后移 print(f\n如果 A 先执行: {result_a}) print(f如果 B 先执行: {result_b}) # 场景 2: 插入 vs 删除 print(\n 场景 2: 插入 vs 删除) print(- * 40) doc Hello World op_a OTOperation(OpType.INSERT, 6, textBeautiful , site_idA, sequence1) op_b OTOperation(OpType.DELETE, 5, deleted_text World, site_idB, sequence1) print(f初始文档: {doc}) print(f用户 A (插入): {op_a}) print(f用户 B (删除): {op_b}) op_a_prime, op_b_prime transform(op_a, op_b) print(f\n变换后:) print(f A {op_a_prime}) print(f B {op_b_prime}) def simulate_collaboration(): 模拟协同编辑过程 print(\n * 65) print( 协同编辑模拟) print( * 65) # 创建两个用户和服务器 doc_a OTDocument(demo, Alice, Hello) doc_b OTDocument(demo, Bob, Hello) server ServerSyncEngine(demo) server.register_client(Alice) server.register_client(Bob) print(f\n初始文档: {doc_a.get_text()}) print(f版本: {doc_a.version}) # Alice 打字 print(\n--- Alice 输入 World ---) alice_op doc_a.insert(5, World) print(fAlice 本地: {doc_a.get_text()}) # 发送到服务器 broadcast server.receive_op(Alice, alice_op) doc_a.acknowledge_op(alice_op.sequence) # Bob 同时打字 print(\n--- Bob 同时输入 !!! ---) bob_op doc_b.insert(5, !!!) print(fBob 本地: {doc_b.get_text()}) # 发送到服务器 broadcast server.receive_op(Bob, bob_op) doc_b.acknowledge_op(bob_op.sequence) # 互相接收 print(\n--- 同步 ---) doc_a.receive_remote_op(bob_op) doc_b.receive_remote_op(alice_op) print(f\n最终 Alice: {doc_a.get_text()}) print(f最终 Bob: {doc_b.get_text()}) print(f一致: {doc_a.get_text() doc_b.get_text()}) if __name__ __main__: demonstrate_transform_rules() simulate_collaboration()八、总结8.1 本讲成果组件文件功能OTOperation​core/ot/operation.py分布式操作定义Transform​core/ot/transform.py四种变换规则SyncEngine​core/ot/sync_engine.pyJupiter 同步模型ServerSyncEngine​core/ot/sync_engine.py服务端同步引擎OTDocument​core/ot/ot_document.pyOT 增强文档引擎8.2 核心知识点OT 的本质通过操作变换保证并发操作的一致性四种变换规则I/I、I/D、D/I、D/D 覆盖所有操作组合Jupiter 模型客户端维护 pending 列表服务端维护全局历史最终一致性无论操作顺序如何所有站点最终结果一致8.3 下一讲预告第3讲WebSocket 实时通信我们将实现真正的网络通信层WebSocket 服务器与客户端操作消息的序列化与广播连接管理与心跳保活断线重连与状态恢复准备好了吗让我们在第3讲再见