OrbitDB 复制（Replication）实战指南：在多 Peer 之间同步数据库

数据库分布式数据库【免费下载链接】orbitdbPeer-to-Peer Databases for the Decentralized Web项目地址https://gitcode.com/gh_mirrors/or/orbitdb点击查看免费下载本篇技术指南以 OrbitDB 官方文档 docs/REPLICATION.md 为主体完整讲解如何让两个 PeerNode.js 进程内运行的两个 IPFS/Helia 实例通过 OrbitDB 的同步协议自动复制数据库数据并深入底层源码剖析复制的工作原理、事件机制与注意事项。读完本文你将能够搭建一个可直接运行的 Peer-to-Peer 数据库复制示例并理解其背后的 pubsub 订阅、heads 交换与最终一致性模型。复制示例两个 Peer 在同一进程内运行官方文档给出了一个最简复制示例两个 Peer 运行在同一个 Node.js 进程中各自拥有独立的 IPFSHelia实例通过 OrbitDB 打开同一个数据库地址后db1 上写入的数据会自动复制到 db2。环境依赖示例依赖以下 npm 包libp2p提供 Peer 之间的网络连接能力helia提供 IPFS 块存储与 libp2p 集成blockstore-level将 IPFS 块持久化到 Level 数据库避免内存存储导致数据丢失orbitdb/coreOrbitDB 核心库。安装命令与 docs/GETTING_STARTED.md 中的说明一致npm i helia orbitdb/core blockstore-levellibp2p 的配置从./config/libp2p.js中导入该配置必须包含gossipsubpubsub 服务因为 OrbitDB 的同步协议依赖 pubsub 主题订阅来发现与通知 Peer。完整的 libp2p 配置示例见 docs/GETTING_STARTED.md 中的Libp2pOptions包含tcp传输、noise加密、yamux多路复用、mdns发现与gossipsubpubsub。完整示例代码import { createLibp2p } from libp2p import { createHelia } from helia import { LevelBlockstore } from blockstore-level import { createOrbitDB } from orbitdb/core import { Libp2pOptions } from ./config/libp2p.js // Our ipfs instances will be connecting over tcp. You can find out more about peer connectivity at https://connectivity.libp2p.io/. const initIPFSInstance () { const blockstore new LevelBlockstore(./ipfs) const libp2p await createLibp2p(Libp2pOptions) return createHelia({ libp2p, blockstore }) } const ipfs1 await initIPFSInstance() const ipfs2 await initIPFSInstance() // The decentralized nature if IPFS can make it slow for peers to find one // another. You can speed up a connection between two peers by dialling-in // to one peer from another. /* await ipfs2.libp2p.save(ipfs1.libp2p.peerId, { multiaddr: ipfs1.libp2p.getMultiaddrs() }) await ipfs2.libp2p.dial(ipfs1.libp2p.peerId) */ const orbitdb1 await createOrbitDB({ ipfs: ipfs1, id: userA, directory: ./orbitdb/1 }) const orbitdb2 await createOrbitDB({ ipfs: ipfs2, id: userB, directory: ./orbitdb/2 }) // This opens a new db. Default db type will be events. const db1 await orbitdb1.open(my-db) // We connect to the first db using its address. This initiates a // synchronization of the heads between db1 and db2. const db2 await orbitdb2.open(db1.address) // We write some data to db1. This will automatically replicated on db2 await db1.add(hello world 1) await db1.add(hello world 2) await db1.add(hello world 3) await db1.add(hello world 4) let db2Updated false // Listen for the connection of ipfs1 to ipfs2. // If we want to listen for connections from ipfs2 to ipfs1, add a join // listener to db1. db2.events.on(join, async (peerId, heads) { // The peerId of the ipfs1 node. console.log(peerId, (await ipfs1.id()).id) }) // Listen for any updates to db2. // If we want to listen for new data on db2, add an update listener to db1. db2.events.on(update, async (entry) { db2Updated true }) // wait for db2 to complete updating. await new Promise((resolve, reject) { setInterval(() { if (db2Updated) { resolve() } }, 1000) }) // Close db1 and its underlying ipfs peer. await db1.close() await orbitdb1.stop() await ipfs1.stop() // Close db2 and its underlying ipfs peer. await db2.close() await orbitdb2.stop() await ipfs2.stop()关键步骤逐段解读创建两个独立的 Helia/IPFS 实例initIPFSInstance()使用LevelBlockstore将块数据持久化到磁盘./ipfs目录。两个实例拥有各自的 libp2p 节点与块存储模拟两个独立的对等节点。创建两个 OrbitDB 实例createOrbitDB({ ipfs, id, directory })为每个 Peer 指定身份id如userA/userB和数据目录。从 src/orbitdb.js 源码可以看到id用于创建身份identitydirectory默认值为./orbitdb用于存放 keystore 等本地数据。db1 创建数据库db2 通过地址打开orbitdb1.open(my-db)创建一个新数据库默认类型为events参见 src/orbitdb.js 中的DefaultDatabaseTypeorbitdb2.open(db1.address)传入 db1 的地址形如/orbitdb/zdpu...打开同一个数据库。这一步会触发两个节点间的 heads 同步。写入数据并自动复制db1.add(...)写入 4 条事件这些数据会自动同步到 db2。监听事件确认复制完成通过db2.events.on(join)与db2.events.on(update)监听节点连接与数据更新。有序清理依次关闭数据库、OrbitDB 实例与 IPFS 实例。关于默认数据库类型 eventsorbitdb1.open(my-db)未指定type因此创建的是events类型数据库。从 src/databases/events.js 源码看add(value)实际上是对addOperation({ op: ADD, key: null, value })的封装——每条事件最终都作为一条 oplog 条目被追加。若需要其他类型可传入{ type: documents }等参数详见 docs/DATABASES.md。复制背后的同步协议Sync Protocol官方文档在结尾提示可查阅 OrbitDB 的同步协议 API。同步协议的实际实现在 src/sync.js其 JSDoc 注释完整描述了协议工作流程这也是理解复制的核心。协议的两个阶段第一阶段初次同步Initial Sync当 Sync 启动时Peer 会订阅日志 ID即数据库地址对应的 pubsub 主题。已在该主题上的 Peer 收到订阅消息后会使用 libp2p 自定义协议协议地址为/orbitdb/heads/log id见 src/sync.js拨号dial新加入的 Peer。建立点对点直连后双方交换各自日志的 heads。交换完成后双方拥有相同的 local state本地状态。第二阶段持续更新Updates初次同步完成后Peer 之间通过最初打开的 pubsub 主题订阅互相通知日志更新。拥有新 heads 的 Peer 将更新后的 heads 发布到 pubsub 主题订阅同一主题的 Peer 收到更新后相应地更新自己的日志状态。最终一致性保证同步协议是最终一致eventually consistent的它保证一旦所有消息都被发送和接收所有 Peer 将观察到相同的日志状态和值。但它不保证消息的接收顺序甚至不保证消息一定能被收到也不保证消息接收的时间。这正是 P2P 去中心化网络下分区容忍partition tolerant设计的体现详见 docs/OPLOG.md。源码中的关键机制从 src/sync.js 源码中可以提炼出以下关键实现细节订阅与取消订阅startSync()中执行pubsub.subscribe(address)src/sync.jsstopSync()中执行pubsub.unsubscribe(address)src/sync.js。address即数据库地址。heads 交换sendHeads()遍历当前日志 heads从存储中取出条目字节流发送给对方src/sync.jsreceiveHeads()逐条解码收到的条目并调用onSynced回调src/sync.js。解密支持接收 heads 时使用log.encryption.replication?.decrypt与log.encryption.data?.decrypt进行解密src/sync.js意味着复制协议原生支持加密数据库。并发串行化使用PQueue({ concurrency: 1 })将所有任务排队串行执行避免并发写入导致的状态竞争src/sync.js。超时控制默认超时为 30 秒DefaultTimeout 30000src/sync.js可通过Sync的timeout参数覆盖。断线清理监听peer:disconnect事件将断开的 Peer 从peers集合中移除否则重连时将无法重新同步 headssrc/sync.js。Database 层如何接入同步在 src/database.js 中Database 创建时即实例化 Syncconst sync await Sync({ ipfs, log, events, onSynced: applyOperation, start: syncAutomatically })onSynced: applyOperation收到远端 heads 时applyOperation调用log.joinEntry(entry)将条目合并进本地 oplog若条目确实新增则触发update事件src/database.js。start: syncAutomatically决定是否自动启动同步该值来自orbitdb.open(address, { sync: true })的sync参数默认true见 src/orbitdb.js。本地写入路径同样会走同步addOperation在log.append(op)之后调用sync.add(entry)将新条目发布到 pubsub 主题src/database.js、src/sync.js从而通知其他 Peer。复制相关事件详解数据库实例db.events暴露了复制相关的事件其定义分散在 src/database.js 与 src/sync.js 的 JSDoc 中事件触发时机回调参数join一个 Peer 完成连接且 heads 交换完成(peerId, heads)对方 PeerID 与收到的 Log 条目数组leave一个 Peer 离开同步协议(peerId)error同步过程发生错误(error)update一条新条目被加入 oplog本地写入或远端复制(entry)新增的 Log 条目官方文档中特别说明了两个方向的监听要点在db2上监听join/update来感知ipfs1 → ipfs2方向的连接与数据更新若想监听反向ipfs2 → ipfs1则需要在db1上添加对应的join/update监听器。测试用例 test/database-replication.test.js 验证了事件语义update事件在添加 1 条操作时恰好触发 1 次添加 4 条操作时恰好触发 4 次且两端db1 与 db2的计数一致——这印证了复制过程中每个条目在写入端与复制端各触发一次update。加速 Peer 连接手动拨号官方文档在示例中以注释形式给出了一段加速连接的代码。由于 IPFS 的去中心化特性Peer 之间互相发现可能较慢通过 libp2p 的 dialling-in 可以显式建立直连await ipfs2.libp2p.save(ipfs1.libp2p.peerId, { multiaddr: ipfs1.libp2p.getMultiaddrs() }) await ipfs2.libp2p.dial(ipfs1.libp2p.peerId)其原理是先把 ipfs1 的 PeerID 与 multiaddr 地址保存到 ipfs2 的 peerStore再主动拨号建立连接。这与测试工具 test/utils/connect-nodes.js 中connectPeers的实现一致Node 环境下peerStore.savedial。说明示例中用的是ipfs2.libp2p.save而在 docs/CONNECTING_PEERS.md 与测试代码中对应 API 是peerStore.save。不同 libp2p 版本 API 可能略有差异以你实际安装的 libp2p 版本为准。若 Peer 分布在不同的网络非 localhost/局域网mDNS 自动发现将失效此时必须通过已知地址手动拨号。更多连接场景Node 与 Node、浏览器与 Node、浏览器与浏览器及中继 relay请参考 docs/CONNECTING_PEERS.md。复制测试如何验证行为仓库中的测试可以作为验证复制行为的直接参考test/database-replication.test.js 验证了数据库跨两个 Peer 复制的核心行为——复制完成后db1.log.iterator()与db2.log.iterator()输出的条目序列完全一致deepStrictEqual(all1, all2)。该测试还覆盖了若干边界场景带延迟的复制写入操作之间插入 1 秒延迟验证复制不受写入节奏影响db2 实例化前先写入db1 先写入数据再创建 db2验证join时的 heads 交换能补齐历史数据收到 head 与持久化条目之间断连模拟收到消息但还没完成持久化时断连的极端情况验证同步依然成功见 test/database-replication.test.js。test/oplog/replicate.test.js 从日志层验证了确定性复制两个 Peer 各自追加 32 条A*/B*条目并通过 pubsub 发布后合并日志的取值顺序固定为A1, B1, A2, B2, ...——这正是 oplog 冲突解决默认 Last Write Wins 写入者 hash 排序保证的确定性结果详见 docs/OPLOG.md。常见问题与注意事项块存储必须持久化Helia 默认使用内存块存储应用结束即丢失数据会导致 OrbitDB 无法再检索复制所需的条目。务必配置LevelBlockstore等持久化存储参考 docs/GETTING_STARTED.md 的Prerequisites一节。pubsub 必须开启libp2p 配置中services.pubsubgossipsub是同步协议运行的前提——订阅主题发现与更新通知都依赖它。同一进程内多 Peer 使用独立目录与身份示例中directory: ./orbitdb/1与./orbitdb/2区分两个 Peer 的本地状态避免数据相互覆盖。等待复制完成复制是异步的示例通过轮询db2Updated标志等待update事件实际应用中更推荐使用Promise 事件监听或类似 test/utils/wait-for.js 的轮询工具。写入权限默认情况下只有数据库创建者可以写入若要让其他 Peer 也能写需要在打开数据库时通过IPFSAccessController({ write: [*] })等访问控制器授权参见 docs/ACCESS_CONTROLLERS.md 与 docs/GETTING_STARTED.md 的Replicating a database一节其中包含完整的双终端运行示例Peer 1 启动后打印 libp2p 地址与 db 地址Peer 2 通过node test.js orbitdb-address libp2p-address拨号接入。网络环境差异mDNS 只适用于本地网络跨网络场景需要手动拨号、公共 bootstrap 节点或 relay 中继详见 docs/CONNECTING_PEERS.md。延伸阅读docs/OPLOG.md复制的数据单元——操作日志oplog的 Merkle-DAG 结构与冲突解决算法理解复制如何做到确定性与防篡改。docs/DATABASES.md不同类型的数据库及其读写 API。docs/CONNECTING_PEERS.md各种网络环境下的 Peer 连接方案。docs/ENCRYPTION.md加密数据库在复制过程中的解密机制。同步协议核心实现src/sync.js数据库接入层src/database.js复制测试test/database-replication.test.js。赞分享数据库分布式数据库【免费下载链接】orbitdbPeer-to-Peer Databases for the Decentralized Web项目地址https://gitcode.com/gh_mirrors/or/orbitdb点击查看免费下载相关推荐Arnis Minecraft 城市生成零基础 5 分钟跑通造一座真实城市Arnis Minecraft 城市生成零基础 5 分钟跑通造一座真实城市 想在 Minecraft 里铺一座街道、建筑、地形都真实的城市吗Arnis 就桌面应用游戏开发GISPouchDB 复制Replication与同步Sync实战深入 CouchDB 多主同步模型与 live/retry 配置PouchDB 复制Replication与同步Sync实战深入 CouchDB 多主同步模型与 live/retry 配置 PouchDB 与 Co数据库数据同步MyBatis-Plus数据同步多数据源之间的实时数据同步方案MyBatis Plus数据同步多数据源之间的实时数据同步方案 引言分布式系统中的数据同步挑战 在现代分布式系统架构中多数据源Multi DataSou后端ORM上一篇多格式支持XML/JSON/Protobuf - 现代HTTP API设计的终极指南下一篇动物行为分析的革命DeepLabCut无标记姿态估计实战指南创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考