第6讲：CRDT 实现

前五讲我们基于 OT 算法构建了协同编辑引擎。OT 虽然强大但它依赖中央服务器来做操作排序和变换。有没有一种方式可以让多个客户端在没有服务器的情况下也能自动达成一致这就是 CRDTConflict-free Replicated Data Types的用武之地。一、CRDT vs OT1.1 核心区别OT 的工作方式 客户端 A ──op1──▶ 服务端 ──op1──▶ 客户端 B 客户端 B ──op2──▶ 服务端 ──op2──▶ 客户端 A 需要服务端做变换保证一致性 CRDT 的工作方式 客户端 A ──op1──▶ 客户端 B 客户端 B ──op2──▶ 客户端 A 无需中心服务端每个客户端独立合并操作 数学保证无论顺序如何最终结果一致1.2 对比表特性OTCRDT中心化​需要中央服务器去中心化离线支持​弱强天然支持算法复杂度​中等较高存储开销​低较高需保留元数据收敛保证​需要正确实现数学保证典型应用​Google DocsFigma, Atom Teletype1.3 本讲实现的 CRDT 类型我们将实现三种 CRDT 1. LWW-Register (Last-Writer-Wins Register) - 最简单的 CRDT - 每个值带时间戳最新的胜出 - 适用于配置、元数据 2. G-Counter (Grow-only Counter) - 只能增加的计数器 - 每个节点维护自己的计数 - 合并时取各节点最大值之和 3. RGA (Replicated Growable Array) - 可增长的数组文本的核心 - 每个字符带唯一 ID - 插入操作天然可交换二、基础 CRDT 类型2.1 LWW-Register# core/crdt/lww_register.py LWW-Register最后写入者胜出的寄存器 from __future__ import annotations from typing import Any, Optional, Tuple from dataclasses import dataclass import time import uuid dataclass class Timestamp: 时间戳带节点 ID保证全局唯一 比较规则先比较时间再比较节点 ID wall_clock: float node_id: str def __lt__(self, other: Timestamp) - bool: if self.wall_clock ! other.wall_clock: return self.wall_clock other.wall_clock return self.node_id other.node_id def __le__(self, other: Timestamp) - bool: return self other or self other def __eq__(self, other: Timestamp) - bool: return (self.wall_clock other.wall_clock and self.node_id other.node_id) def to_dict(self) - dict: return { wall_clock: self.wall_clock, node_id: self.node_id } classmethod def from_dict(cls, data: dict) - Timestamp: return cls( wall_clockdata[wall_clock], node_iddata[node_id] ) classmethod def now(cls, node_id: str) - Timestamp: return cls(wall_clocktime.time(), node_idnode_id) class LWWRegister: LWW-Register 每个副本保存 (值, 时间戳) 对。 合并时取时间戳最大的值。 def __init__(self, node_id: str, initial_value: Any None): self.node_id node_id self.value initial_value self.timestamp Timestamp.now(node_id) if initial_value else None def set(self, value: Any): 设置值 Args: value: 新值 self.value value self.timestamp Timestamp.now(self.node_id) def get(self) - Any: 获取当前值 return self.value def merge(self, other: LWWRegister): 合并另一个副本 取时间戳更大的值。 Args: other: 另一个副本 if other.timestamp and self.timestamp: if other.timestamp self.timestamp: self.value other.value self.timestamp other.timestamp elif other.timestamp and not self.timestamp: self.value other.value self.timestamp other.timestamp def to_dict(self) - dict: return { value: self.value, timestamp: self.timestamp.to_dict() if self.timestamp else None, node_id: self.node_id } classmethod def from_dict(cls, data: dict) - LWWRegister: reg cls(node_iddata[node_id]) if data[timestamp]: reg.timestamp Timestamp.from_dict(data[timestamp]) reg.value data[value] return reg def __repr__(self) - str: return fLWWRegister({self.value} {self.timestamp})2.2 G-Counter# core/crdt/gcounter.py G-Counter只能增加的计数器 from __future__ import annotations from typing import Dict from dataclasses import dataclass class GCounter: G-Counter (Grow-only Counter) 每个节点维护自己的计数。 合并时取各节点计数的最大值之和。 特性 - 只能增加 - 天然支持并发 - 最终一致 def __init__(self, node_id: str): self.node_id node_id self.counts: Dict[str, int] {node_id: 0} def increment(self, amount: int 1): 增加计数 Args: amount: 增加量 self.counts[self.node_id] self.counts.get(self.node_id, 0) amount def get(self) - int: 获取总值 return sum(self.counts.values()) def merge(self, other: GCounter): 合并另一个计数器 对每个节点取最大值。 for node_id, count in other.counts.items(): self.counts[node_id] max( self.counts.get(node_id, 0), count ) def to_dict(self) - dict: return { node_id: self.node_id, counts: dict(self.counts) } classmethod def from_dict(cls, data: dict) - GCounter: counter cls(node_iddata[node_id]) counter.counts data[counts] return counter def __repr__(self) - str: return fGCounter({self.get()})三、RGA文本的核心 CRDT3.1 RGA 原理RGA 的核心思想 每个字符都有一个唯一的 ID (node_id, seq)。 字符之间通过前驱指针链接形成链表。 插入操作 1. 为新字符生成唯一 ID 2. 找到插入位置的前驱字符 3. 将新字符插入到前驱之后 删除操作墓碑 1. 标记字符为已删除墓碑 2. 不真正删除保留位置信息 合并规则 1. 如果两个字符 ID 相同保留一个 2. 如果两个字符插入到同一位置按 ID 排序 3. 删除标记可以安全合并 示例 初始: He (H:id1, e:id2) 节点 A 插入 l 到位置 2: Hel (H:id1, e:id2, l:id3_A) 节点 B 同时插入 x 到位置 2: Hex (H:id1, e:id2, x:id3_B) 合并后按 ID 排序: Helx 或 Hexl取决于 ID 大小3.2 RGA 实现# core/crdt/rga.py RGA (Replicated Growable Array) 基于链表的可增长数组用于表示协同编辑的文本。 from __future__ import annotations from typing import List, Optional, Tuple, Dict from dataclasses import dataclass, field import uuid dataclass class RGAPosition: RGA 中的位置标识 每个字符的唯一标识。 node_id: str # 创建该字符的节点 seq: int # 节点内的序号 def __lt__(self, other: RGAPosition) - bool: 按 (seq, node_id) 排序 if self.seq ! other.seq: return self.seq other.seq return self.node_id other.node_id def __eq__(self, other: RGAPosition) - bool: return self.node_id other.node_id and self.seq other.seq def __hash__(self) - int: return hash((self.node_id, self.seq)) def to_dict(self) - dict: return {node_id: self.node_id, seq: self.seq} classmethod def from_dict(cls, data: dict) - RGAPosition: return cls(node_iddata[node_id], seqdata[seq]) dataclass class RGANode: RGA 链表节点 代表一个字符或已删除的字符墓碑。 position: RGAPosition # 唯一位置标识 value: str # 字符值 deleted: bool False # 是否已删除墓碑 next_pos: Optional[RGAPosition] None # 后继节点位置 def to_dict(self) - dict: return { position: self.position.to_dict(), value: self.value, deleted: self.deleted, next_pos: self.next_pos.to_dict() if self.next_pos else None } classmethod def from_dict(cls, data: dict) - RGANode: return cls( positionRGAPosition.from_dict(data[position]), valuedata[value], deleteddata.get(deleted, False), next_posRGAPosition.from_dict(data[next_pos]) if data.get(next_pos) else None ) class RGA: RGA (Replicated Growable Array) 基于链表的 CRDT 文本类型。 特性 - 插入操作天然可交换 - 删除使用墓碑标记 - 合并后自动收敛 def __init__(self, node_id: str): self.node_id node_id # 节点映射: position - RGANode self.nodes: Dict[RGAPosition, RGANode] {} # 头节点虚拟根节点 self.head RGAPosition(node_id__head__, seq-1) # 序号计数器 self.seq_counter 0 # 墓碑计数用于统计 self.tombstone_count 0 # ---------- 本地操作 ---------- def insert(self, index: int, value: str) - List[RGANode]: 在指定位置插入字符串 Args: index: 插入位置 value: 要插入的字符串 Returns: 创建的节点列表 if index 0: index 0 # 找到插入位置的前驱 prev_pos self._find_position(index) created_nodes [] for ch in value: # 生成新节点 self.seq_counter 1 pos RGAPosition(node_idself.node_id, seqself.seq_counter) node RGANode( positionpos, valuech, next_posNone ) # 插入到链表 if prev_pos in self.nodes: node.next_pos self.nodes[prev_pos].next_pos self.nodes[prev_pos].next_pos pos self.nodes[pos] node created_nodes.append(node) prev_pos pos return created_nodes def delete(self, index: int, length: int 1) - List[RGAPosition]: 删除指定位置的字符 Args: index: 起始位置 length: 删除长度 Returns: 被删除节点的位置列表 deleted [] current self.head # 遍历到起始位置 pos_idx 0 while current: if current in self.nodes: node self.nodes[current] if not node.deleted: if pos_idx index and pos_idx index length: node.deleted True self.tombstone_count 1 deleted.append(current) pos_idx 1 # 移动到下一个 if current in self.nodes: current self.nodes[current].next_pos else: break return deleted # ---------- 查询 ---------- def get_text(self) - str: 获取可见文本 chars [] current self.head while current: if current in self.nodes: node self.nodes[current] if not node.deleted: chars.append(node.value) current node.next_pos else: break return .join(chars) def get_length(self) - int: 获取可见文本长度 return len(self.get_text()) def _find_position(self, index: int) - RGAPosition: 找到第 index 个可见字符的前驱位置 Args: index: 目标索引 Returns: 前驱节点的位置 current self.head visible_count 0 while current: if current in self.nodes: node self.nodes[current] if not node.deleted: if visible_count index: return current visible_count 1 next_pos node.next_pos if next_pos and next_pos in self.nodes: current next_pos else: break else: break # 返回最后一个节点 return current # ---------- 合并 ---------- def merge(self, other: RGA): 合并另一个 RGA 副本 合并规则 1. 相同位置的节点保留一个 2. 删除标记合并任一标记删除即删除 3. 链表结构合并 # 合并所有节点 for pos, node in other.nodes.items(): if pos in self.nodes: # 已存在合并删除标记 if node.deleted: self.nodes[pos].deleted True else: # 新节点直接添加 self.nodes[pos] RGANode( positionpos, valuenode.value, deletednode.deleted, next_posnode.next_pos ) # 更新墓碑计数 self.tombstone_count sum( 1 for n in self.nodes.values() if n.deleted ) # ---------- 序列化 ---------- def to_dict(self) - dict: return { node_id: self.node_id, seq_counter: self.seq_counter, nodes: { pos.to_dict(): node.to_dict() for pos, node in self.nodes.items() }, head: self.head.to_dict() } classmethod def from_dict(cls, data: dict) - RGA: rga cls(node_iddata[node_id]) rga.seq_counter data[seq_counter] for pos_data, node_data in data[nodes].items(): pos RGAPosition.from_dict( eval(pos_data) if isinstance(pos_data, str) else pos_data ) rga.nodes[pos] RGANode.from_dict(node_data) return rga def garbage_collect(self, threshold: float 0.3): 垃圾回收清理墓碑 当墓碑比例超过阈值时重建链表。 Args: threshold: 墓碑比例阈值 total len(self.nodes) if total 0: return tombstone_ratio self.tombstone_count / total if tombstone_ratio threshold: return # 重建只保留可见节点 text self.get_text() self.nodes.clear() self.tombstone_count 0 self.seq_counter 0 # 重新插入 self.insert(0, text) def __repr__(self) - str: return fRGA({self.get_text()}, nodes{len(self.nodes)}, tombstones{self.tombstone_count})四、CRDT 文档引擎4.1 CRDTDocument# core/crdt/crdt_document.py 基于 CRDT 的文档引擎 from __future__ import annotations from typing import List, Optional, Dict from dataclasses import dataclass import time import logging from core.crdt.rga import RGA, RGANode, RGAPosition from core.crdt.lww_register import LWWRegister from core.crdt.gcounter import GCounter logger logging.getLogger(__name__) dataclass class DocumentSnapshot: 文档快照 text: str version: int timestamp: float node_count: int def to_dict(self) - dict: return { text: self.text, version: self.version, timestamp: self.timestamp, node_count: self.node_count } class CRDTDocument: 基于 CRDT 的文档引擎 使用 RGA 表示文本内容LWWRegister 管理元数据。 天然支持去中心化同步。 def __init__(self, document_id: str, node_id: str): self.document_id document_id self.node_id node_id # 文本内容RGA self.content RGA(node_id) # 元数据LWWRegister self.title LWWRegister(node_id, Untitled) self.author LWWRegister(node_id, ) # 操作计数 self.op_count GCounter(node_id) # 版本号 self.version 0 # 变更回调 self.on_change: Optional[callable] None # ---------- 文本操作 ---------- def insert(self, index: int, text: str) - List[RGANode]: 插入文本 Args: index: 位置 text: 文本 Returns: 创建的节点 nodes self.content.insert(index, text) self.op_count.increment() self.version 1 if self.on_change: self.on_change(insert, index, text) return nodes def delete(self, index: int, length: int 1) - List[RGAPosition]: 删除文本 Args: index: 起始位置 length: 删除长度 Returns: 被删除的位置 positions self.content.delete(index, length) self.op_count.increment() self.version 1 if self.on_change: self.on_change(delete, index, length) return positions # ---------- 元数据操作 ---------- def set_title(self, title: str): 设置标题 self.title.set(title) self.version 1 def set_author(self, author: str): 设置作者 self.author.set(author) self.version 1 # ---------- 查询 ---------- def get_text(self) - str: 获取文本 return self.content.get_text() def get_length(self) - int: 获取长度 return self.content.get_length() def get_title(self) - str: 获取标题 return self.title.get() or Untitled def get_op_count(self) - int: 获取操作数 return self.op_count.get() # ---------- 合并 ---------- def merge(self, other: CRDTDocument): 合并另一个文档副本 Args: other: 另一个文档 # 合并文本内容 self.content.merge(other.content) # 合并元数据 self.title.merge(other.title) self.author.merge(other.author) # 合并操作计数 self.op_count.merge(other.op_count) # 更新版本号 self.version max(self.version, other.version) 1 logger.info(fMerged document: version{self.version}) # ---------- 快照 ---------- def snapshot(self) - DocumentSnapshot: 创建快照 return DocumentSnapshot( textself.get_text(), versionself.version, timestamptime.time(), node_countlen(self.content.nodes) ) # ---------- 序列化 ---------- def to_dict(self) - dict: return { document_id: self.document_id, node_id: self.node_id, content: self.content.to_dict(), title: self.title.to_dict(), author: self.author.to_dict(), op_count: self.op_count.to_dict(), version: self.version } classmethod def from_dict(cls, data: dict) - CRDTDocument: doc cls( document_iddata[document_id], node_iddata[node_id] ) doc.content RGA.from_dict(data[content]) doc.title LWWRegister.from_dict(data[title]) doc.author LWWRegister.from_dict(data[author]) doc.op_count GCounter.from_dict(data[op_count]) doc.version data.get(version, 0) return doc五、CRDT 同步5.1 同步协议# core/crdt/sync.py CRDT 同步协议 from __future__ import annotations from typing import List, Optional, Dict from dataclasses import dataclass import time import logging from core.crdt.crdt_document import CRDTDocument from core.crdt.rga import RGA logger logging.getLogger(__name__) dataclass class SyncMessage: 同步消息 node_id: str document_id: str changes: dict timestamp: float def to_dict(self) - dict: return { node_id: self.node_id, document_id: self.document_id, changes: self.changes, timestamp: self.timestamp } class CRDTSyncEngine: CRDT 同步引擎 管理 CRDT 文档的同步。 由于 CRDT 的数学保证只需要交换状态即可。 def __init__(self, document: CRDTDocument): self.document document # 已知的其他节点状态 self.known_nodes: Dict[str, int] {} # node_id - last_version def generate_sync_message(self) - SyncMessage: 生成同步消息 包含完整的文档状态。 return SyncMessage( node_idself.document.node_id, document_idself.document.document_id, changesself.document.to_dict(), timestamptime.time() ) def apply_sync_message(self, message: SyncMessage) - bool: 应用同步消息 Args: message: 同步消息 Returns: 是否有变更 if message.node_id self.document.node_id: return False # 忽略自己的消息 # 重建文档 remote_doc CRDTDocument.from_dict(message.changes) # 合并 before_version self.document.version self.document.merge(remote_doc) changed self.document.version before_version if changed: self.known_nodes[message.node_id] remote_doc.version logger.info(fApplied sync from {message.node_id}: fversion {before_version} → {self.document.version}) return changed def get_sync_status(self) - dict: 获取同步状态 return { node_id: self.document.node_id, document_version: self.document.version, known_nodes: len(self.known_nodes), text_length: self.document.get_length() }六、演示6.1 CRDT 收敛演示# examples/crdt_demo.py CRDT 收敛性演示 import logging import sys import threading import time sys.path.insert(0, ..) from core.crdt.crdt_document import CRDTDocument from core.crdt.sync import CRDTSyncEngine logging.basicConfig(levellogging.INFO) def demonstrate_crdt_convergence(): 演示 CRDT 的收敛性 print( * 78) print( CRDT 收敛性演示) print( * 78) # 创建三个节点的文档 doc_a CRDTDocument(doc-1, Alice) doc_b CRDTDocument(doc-1, Bob) doc_c CRDTDocument(doc-1, Carol) print(f\n初始状态:) print(f Alice: {doc_a.get_text()}) print(f Bob: {doc_b.get_text()}) print(f Carol: {doc_c.get_text()}) # 三个节点同时编辑 print(\n--- 三人同时编辑 ---) # Alice 输入 Hello doc_a.insert(0, Hello) print(f Alice 输入: Hello) # Bob 输入 World doc_b.insert(0, World) print(f Bob 输入: World) # Carol 输入 CRDT doc_c.insert(0, CRDT) print(f Carol 输入: CRDT) print(f\n同步前:) print(f Alice: {doc_a.get_text()}) print(f Bob: {doc_b.get_text()}) print(f Carol: {doc_c.get_text()}) # 两两同步 print(\n--- 开始同步 ---) sync_a CRDTSyncEngine(doc_a) sync_b CRDTSyncEngine(doc_b) sync_c CRDTSyncEngine(doc_c) # Alice ↔ Bob msg_a sync_a.generate_sync_message() sync_b.apply_sync_message(msg_a) msg_b sync_b.generate_sync_message() sync_a.apply_sync_message(msg_b) print( Alice ↔ Bob 同步完成) # Bob ↔ Carol msg_b sync_b.generate_sync_message() sync_c.apply_sync_message(msg_b) msg_c sync_c.generate_sync_message() sync_b.apply_sync_message(msg_c) print( Bob ↔ Carol 同步完成) # Carol ↔ Alice msg_c sync_c.generate_sync_message() sync_a.apply_sync_message(msg_c) msg_a sync_a.generate_sync_message() sync_c.apply_sync_message(msg_a) print( Carol ↔ Alice 同步完成) print(f\n同步后应完全一致:) print(f Alice: {doc_a.get_text()}) print(f Bob: {doc_b.get_text()}) print(f Carol: {doc_c.get_text()}) consistent (doc_a.get_text() doc_b.get_text() doc_c.get_text()) print(f\n一致性: {✅ 一致! if consistent else ❌ 不一致!}) def demonstrate_crdt_vs_ot(): 演示 CRDT vs OT 的区别 print(\n * 78) print(⚔️ CRDT vs OT 对比演示) print( * 78) # CRDT 方式 print(\n CRDT 方式去中心化:) alice CRDTDocument(doc-2, Alice) bob CRDTDocument(doc-2, Bob) alice.insert(0, AB) bob.insert(0, XY) # 直接合并 alice.merge(bob) bob.merge(alice) print(f Alice: {alice.get_text()}) print(f Bob: {bob.get_text()}) print(f 一致: {alice.get_text() bob.get_text()}) print(f 无需中心服务器!) # 对比 OT需要服务端 print(\n OT 方式需要中心服务端:) print( 需要服务端做操作变换) print( 需要维护操作顺序) print( 需要处理确认/重试) def demonstrate_garbage_collection(): 演示墓碑垃圾回收 print(\n * 78) print(️ RGA 墓碑垃圾回收) print( * 78) doc CRDTDocument(doc-3, Demo) # 输入长文本 doc.insert(0, This is a long text that we will edit many times) print(f\n初始: {doc.get_text()}) print(f节点数: {len(doc.content.nodes)}) # 多次编辑产生墓碑 print(\n--- 多次编辑 ---) for i in range(5): doc.delete(0, 5) doc.insert(0, XXXXX) print(f 编辑 {i1}: {doc.get_text()}) print(f\n编辑后:) print(f 节点数: {len(doc.content.nodes)}) print(f 墓碑数: {doc.content.tombstone_count}) print(f 墓碑比例: {doc.content.tombstone_count/len(doc.content.nodes)*100:.1f}%) # 垃圾回收 print(\n--- 垃圾回收 ---) doc.content.garbage_collect(threshold0.2) print(f 回收后节点数: {len(doc.content.nodes)}) print(f 回收后墓碑数: {doc.content.tombstone_count}) print(f 文本不变: {doc.get_text()}) def main(): demonstrate_crdt_convergence() demonstrate_crdt_vs_ot() demonstrate_garbage_collection() if __name__ __main__: main()七、测试7.1 CRDT 测试# tests/test_crdt.py import pytest from core.crdt.lww_register import LWWRegister, Timestamp from core.crdt.gcounter import GCounter from core.crdt.rga import RGA, RGAPosition, RGANode from core.crdt.crdt_document import CRDTDocument from core.crdt.sync import CRDTSyncEngine class TestLWWRegister: LWW-Register 测试 def test_set_and_get(self): reg LWWRegister(node-1) reg.set(hello) assert reg.get() hello def test_merge(self): reg1 LWWRegister(node-1, hello) reg2 LWWRegister(node-2, world) # 确保 reg2 的时间戳更新 import time time.sleep(0.001) reg2.set(world) reg1.merge(reg2) assert reg1.get() world class TestGCounter: G-Counter 测试 def test_increment(self): counter GCounter(node-1) counter.increment() counter.increment(3) assert counter.get() 4 def test_merge(self): c1 GCounter(node-1) c2 GCounter(node-2) c1.increment(5) c2.increment(3) c1.merge(c2) assert c1.get() 8 class TestRGA: RGA 测试 def test_insert_and_get(self): rga RGA(node-1) rga.insert(0, Hello) assert rga.get_text() Hello def test_delete(self): rga RGA(node-1) rga.insert(0, Hello) rga.delete(0, 2) assert rga.get_text() llo def test_merge_convergence(self): rga_a RGA(Alice) rga_b RGA(Bob) rga_a.insert(0, AB) rga_b.insert(0, XY) rga_a.merge(rga_b) rga_b.merge(rga_a) assert rga_a.get_text() rga_b.get_text() class TestCRDTDocument: CRDT 文档测试 def test_basic_operations(self): doc CRDTDocument(doc-1, Alice) doc.insert(0, Hello) assert doc.get_text() Hello doc.delete(0, 2) assert doc.get_text() llo def test_merge_convergence(self): doc_a CRDTDocument(doc-1, Alice) doc_b CRDTDocument(doc-1, Bob) doc_c CRDTDocument(doc-1, Carol) # 三人同时编辑 doc_a.insert(0, Hello) doc_b.insert(0, World) doc_c.insert(0, CRDT) # 两两合并 doc_a.merge(doc_b) doc_a.merge(doc_c) doc_b.merge(doc_a) doc_b.merge(doc_c) doc_c.merge(doc_a) doc_c.merge(doc_b) assert doc_a.get_text() doc_b.get_text() assert doc_b.get_text() doc_c.get_text() if __name__ __main__: pytest.main([__file__, -v])八、总结8.1 本讲成果组件文件功能LWWRegister​core/crdt/lww_register.py最后写入者胜出寄存器GCounter​core/crdt/gcounter.py只能增加的计数器RGA​core/crdt/rga.py可增长数组文本核心CRDTDocument​core/crdt/crdt_document.py基于 CRDT 的文档引擎CRDTSyncEngine​core/crdt/sync.pyCRDT 同步引擎8.2 核心知识点CRDT 的本质通过数据结构设计保证最终一致性无需中心服务器RGA 的工作原理链表 唯一 ID 墓碑标记合并的数学保证无论操作顺序如何最终结果一致墓碑机制删除不真删保留位置信息垃圾回收定期清理墓碑控制内存增长8.3 下一讲预告第7讲选区同步与光标展示我们将实现多人编辑时最重要的视觉反馈远程光标渲染选区同步光标颜色管理平滑动画过渡准备好了吗让我们在第7讲再见开发之余的小工具推荐​处理 Base64、JWT 解析、JSON 格式化、Crontab 计算、PDF 合并压缩这些碎片需求我常用一个纯前端本地工具箱zz365.top子页 PDF 大师PDF 大师 - zz365工具箱。所有计算在浏览器完成文件不上传服务器关页即清。免费、无登录、无广告适合开发者当常驻标签页。