单实例MongoDB数据同步：基于Change Stream的增量同步方案-北京尧图网络

1. 为什么单实例 MongoDB 之间也需要数据同步很多人一听到数据同步第一反应就是副本集、分片集群那一套。但实际工作中我们遇到的场景往往没那么高大上——两台独立的服务器各自跑着一个单实例 MongoDB现在需要把 A 库里的部分数据搬到 B 库或者让两边保持某种程度的同步。这种需求在数据迁移、灰度环境搭建、跨机房数据归集、测试环境数据回灌等场景里非常常见。我最早接触这类需求是在一个数据归集项目里边缘节点上跑着单实例 MongoDB用来暂存采集到的数据中心节点也是单实例需要定期把边缘节点的数据汇总上来。当时第一反应是写脚本导出再导入但数据量一上来mongodumpmongorestore那套就顶不住了而且没法做增量。后来才认真研究了 MongoDB 自带的 Change Stream 机制也就是标题里说的 mongo shake——这里其实是社区里对Change Stream的口语化叫法因为它的核心就是监听数据变化流读起来像 shake。先把概念说清楚Change Stream 是 MongoDB 3.6 引入的特性它建立在 oplog 之上允许应用程序实时订阅集合、数据库或整个部署上的数据变更。每条变更事件会告诉你发生了什么操作、操作在哪个集合、文档的 key 是什么、变更前后的内容是什么。有了它你就能在单实例之间搭一条数据管道把源库的写入实时搬运到目标库。这个方案适合谁我认为三类人最需要一是做数据迁移但不想停机的人二是要搭轻量级跨库同步但不想上副本集的人三是想理解 MongoDB 变更捕获机制、为后续做 CDC变更数据捕获打基础的开发者。哪怕你只是想搞清楚 oplog 和 Change Stream 的关系这篇内容也能帮你把链路走通。需要提前说明一个硬性前提Change Stream 依赖 oplog而 oplog 只在副本集模式下存在。单实例 MongoDB 默认是没有 oplog 的。所以标题里说的单实例之间同步严格来讲需要把源库以单节点副本集的方式启动——它物理上还是一台机器、一个实例但逻辑上具备副本集的能力从而产生 oplog。这是整个方案能不能跑起来的关键后面会详细讲怎么配。2. 方案整体设计与核心思路拆解2.1 为什么选 Change Stream 而不是定时轮询做数据同步最朴素的做法是定时轮询每隔几秒查一次源库把updatedAt大于上次同步时间的文档捞出来写进目标库。这个方案我早期用过问题很明显。第一删除操作捕获不到——文档被删了你轮询时它已经不在集合里了目标库的对应文档就成了僵尸数据。第二时间戳字段不可靠——如果业务代码忘了更新updatedAt或者多台机器时钟不一致就会漏数据或重复同步。第三轮询有延迟且浪费资源——高频轮询压数据库低频轮询延迟大很难两全。Change Stream 从根上解决了这几个问题。它直接读 oplog插入、更新、删除、替换、drop 集合等操作全都能捕获而且事件是推过来的源库一有写入订阅端几乎立刻收到延迟通常在毫秒级。更重要的是它带resume token恢复令牌程序崩溃重启后可以从上次断开的位置继续不会丢事件。这一点对于生产环境的可靠性至关重要。2.2 单节点副本集物理单机逻辑集群前面提到Change Stream 必须有 oplog。那怎么让单实例产生 oplog答案是以副本集模式启动但只配置一个成员。具体做法是在启动参数里加上--replSet指定一个副本集名字然后执行一次rs.initiate()。这样这个实例就变成了一个单节点副本集它会维护自己的 oplogChange Stream 也就能工作了。这个设计的好处是你不需要真的部署三节点集群一台机器、一个进程就能搞定运维成本和单实例几乎一样。代价是失去了副本集的高可用——这台机器挂了就是挂了没有自动故障转移。但对于数据同步这种源库本来就在跑的场景源库的高可用是它自己的事同步程序只负责读不改变源库的部署形态。注意如果你的源库已经在以单实例模式运行且不能重启那就没法直接开 Change Stream。这种情况下要么安排一次重启窗口改成单节点副本集要么退回到轮询方案。这一点必须在动手前确认清楚。2.3 同步管道的三段式结构整个同步链路我习惯拆成三段监听端Watcher、转换层Transformer、写入端Writer。监听端负责在源库上开一个 Change Stream拿到原始变更事件。转换层负责把事件翻译成目标库能执行的操作——比如把insert事件转成目标库的insertOne把delete事件转成deleteOne。写入端负责把转换后的操作落到目标库并处理冲突比如目标库已经有同_id的文档。为什么要拆成三段而不是一把梭因为每段的失败模式不一样拆开后好排查。监听端断了通常是网络或权限问题写入端报错多半是唯一键冲突或字段类型不匹配。分开之后日志能精确定位到是哪一段出的问题而不是笼统地报一个同步失败。2.4 幂等性同步程序的命根子任何同步方案都必须回答一个问题同一条变更被处理两次会怎样网络抖动、程序重启、resume token 回退都可能导致重复消费。如果写入端不做幂等处理目标库就会出现重复数据或者报错中断。我的做法是以_id为准做 upsert。对于 insert 和 update 事件统一用replaceOne({_id: ...}, doc, {upsert: true})写入。这样无论同一条变更来几次结果都一样。对于 delete 事件用deleteOne({_id: ...})删两次也不会有副作用。这个策略牺牲了一点性能replace 比 insert 慢但换来的是随便重试都不怕的安心感非常值得。3. 核心细节解析与实操要点3.1 把单实例改造成单节点副本集这是整个方案的地基必须做对。假设你的 MongoDB 是用配置文件启动的找到mongod.conf在里边加上副本集配置replication: replSetName: rs0然后重启 mongod。重启完成后用 mongosh 连上去执行初始化rs.initiate({ _id: rs0, members: [ { _id: 0, host: 127.0.0.1:27017 } ] })执行完等几秒用rs.status()看一下stateStr显示PRIMARY就说明成功了。这时候再执行db.getSiblingDB(local).oplog.rs.stats()能看到 oplog 集合已经存在说明地基打好了。这里有个容易踩的坑host字段填的地址必须是订阅端能访问到的地址。如果你填了127.0.0.1而订阅程序跑在另一台机器上那它连上来之后会被副本集告知主节点在 127.0.0.1然后就连不上了。所以跨机同步时host 要填真实的、可被外部访问的 IP 或域名。这个坑我踩过一次排查了半天才发现是副本集配置里的地址问题。3.2 权限配置别用 root 裸奔Change Stream 需要读 oplog 的权限。生产环境绝对不要用 root 账号跑同步程序应该单独建一个最小权限账号。在源库上执行use admin db.createUser({ user: sync_reader, pwd: your_strong_password, roles: [ { role: read, db: your_source_db }, { role: read, db: local } ] })local库的读权限是读 oplog 必需的your_source_db的读权限是读业务数据必需的。目标库那边则要给一个能读写的账号角色用readWrite就够了。提示有些团队图省事直接给clusterMonitor或readAnyDatabase权限给大了。最小权限原则在同步场景同样适用万一同步程序的凭证泄露损失可控。3.3 Change Stream 的打开方式与过滤打开 Change Stream 的代码很简洁以 Python 为例from pymongo import MongoClient client MongoClient(mongodb://sync_reader:pwdsource_host:27017/?replicaSetrs0) db client[your_source_db] collection db[your_collection] pipeline [ {$match: {operationType: {$in: [insert, update, replace, delete]}}} ] stream collection.watch(pipeline, full_documentupdateLookup)几个关键点值得展开。pipeline参数允许你在服务端就过滤掉不关心的事件比如只同步某几种操作类型或者只同步某个字段满足条件的文档。服务端过滤能大幅减少网络传输比拉回来再过滤高效得多。full_documentupdateLookup这个参数非常重要。默认情况下update 事件只告诉你哪个文档被改了、改了哪些字段但不给完整文档。设成updateLookup后MongoDB 会额外去查一次当前完整文档附在事件里。这样写入端就能直接 replace不用自己拼字段。代价是每次 update 多一次查询但省去了大量拼接逻辑我认为很划算。resume_after是断点续传的关键。每次处理完一个事件把事件的_id存下来可以存本地文件也可以存目标库的一个专用集合。程序重启时把它传给watch()就能从断点继续。这个机制必须实现否则程序一崩就得全量重来。3.4 事件结构长什么样理解事件结构是写好转换层的前提。一条典型的 insert 事件大概是这样{ _id: { _data: 8264A1B2C3... }, operationType: insert, clusterTime: Timestamp(1700000000, 1), fullDocument: { _id: abc123, name: foo, age: 30 }, ns: { db: your_source_db, coll: your_collection }, documentKey: { _id: abc123 } }operationType决定你要做什么操作fullDocument是文档内容delete 事件没有这个字段documentKey是文档主键ns告诉你来自哪个库哪个集合。转换层要做的就是根据operationType分派insert/update/replace 走 upsertdelete 走 deleteOne。有个细节要注意update 事件的fullDocument只有在设了updateLookup时才有而且它反映的是事件被处理那一刻的文档状态不是变更发生那一刻的状态。如果同一个文档在极短时间内被改了多次你拿到的可能是最新版本。这在绝大多数场景下没问题但如果你的业务对中间状态敏感就得改用full_document_before_changeMongoDB 6.0 支持来拿变更前的快照。4. 实操过程与核心环节实现4.1 环境准备清单动手前先把这些东西备齐能省掉大量来回折腾的时间项目源库要求目标库要求MongoDB 版本3.6 以上建议 4.43.6 以上部署模式单节点副本集单实例即可账号权限read local 读readWrite网络订阅端能访问源库端口订阅端能访问目标库端口驱动pymongo 3.9 / mongo-go-driver 1.4同左版本这块我建议源库至少 4.4。3.6 虽然支持 Change Stream但早期版本在 resume token 的稳定性和updateLookup的性能上都有欠缺4.4 之后成熟很多。如果条件允许直接上 5.0 或 6.0full_document_before_change这类新特性在复杂同步场景里很有用。4.2 完整同步程序骨架下面给一个可以直接改改就用的 Python 骨架。我把它写成一个类方便你扩展import time from pymongo import MongoClient, ReplaceOne, DeleteOne from pymongo.errors import PyMongoError class MongoSyncer: def __init__(self, source_uri, target_uri, db_name, coll_name, token_file): self.source MongoClient(source_uri) self.target MongoClient(target_uri) self.db_name db_name self.coll_name coll_name self.token_file token_file self.src_coll self.source[db_name][coll_name] self.tgt_coll self.target[db_name][coll_name] def load_token(self): try: with open(self.token_file, r) as f: return eval(f.read()) except FileNotFoundError: return None def save_token(self, token): with open(self.token_file, w) as f: f.write(repr(token)) def build_op(self, event): op_type event[operationType] key event[documentKey] if op_type in (insert, update, replace): doc event.get(fullDocument) if doc is None: return None return ReplaceOne(key, doc, upsertTrue) elif op_type delete: return DeleteOne(key) return None def run(self): token self.load_token() pipeline [{$match: { operationType: {$in: [insert, update, replace, delete]} }}] while True: try: kwargs {full_document: updateLookup} if token: kwargs[resume_after] token with self.src_coll.watch(pipeline, **kwargs) as stream: for event in stream: op self.build_op(event) if op is not None: self.tgt_coll.bulk_write([op], orderedFalse) self.save_token(event[_id]) except PyMongoError as e: print(fstream error: {e}, retrying in 3s) time.sleep(3) if __name__ __main__: syncer MongoSyncer( source_urimongodb://sync_reader:pwdsource_host:27017/?replicaSetrs0, target_urimongodb://writer:pwdtarget_host:27017/, db_nameyour_source_db, coll_nameyour_collection, token_file./resume_token.txt ) syncer.run()这段代码有几个设计取舍值得说。用bulk_write而不是单条写是为了将来扩展成批量处理时不用改结构现在一次写一条也没问题。orderedFalse表示不要求顺序执行遇到单条失败不影响其他条配合幂等写入很合适。外层while True 内层with watch的结构是为了在流断开时自动重连重连时用保存的 token 续上。4.3 resume token 的存储选择token 存哪里是个需要想清楚的问题。存本地文件最简单但如果同步程序部署在容器里容器一重建文件就没了token 丢失意味着要么全量重来要么从当前时间点开始会漏数据。所以生产环境我更推荐把 token 存到目标库的一个专用集合里def save_token(self, token): self.target[sync_meta][resume_tokens].update_one( {_id: f{self.db_name}.{self.coll_name}}, {$set: {token: token, updated_at: time.time()}}, upsertTrue )这样 token 和同步数据在同一个库备份、迁移都一起走不会出现数据在但断点丢了的尴尬。代价是每次处理事件多一次写操作但相比数据同步本身的开销这点写入可以忽略。4.4 首次全量后续增量的衔接Change Stream 只能捕获打开流之后发生的变更。那打开流之前就存在的数据怎么办答案是先做一次全量同步再开流。顺序不能反否则全量期间的新写入会丢。标准流程是第一步记录当前时间点T0第二步把源库全量数据 dump 到目标库第三步从T0开始打开 Change Stream。这样全量覆盖了T0之前的数据增量覆盖了T0之后的数据中间无缝衔接。不过这里有个微妙的边界问题全量 dump 本身要花时间如果 dump 过程中源库有写入这些写入既不在 dump 的快照里取决于 dump 的一致性又可能早于T0。稳妥的做法是用start_at_operation_time参数把流的起点精确设到全量快照的那个时间点。MongoDB 的mongodump在副本集模式下会记录oplog位置可以拿来对齐。这个细节比较深如果数据量不大简单做法是全量期间停写或者接受极小概率的重复反正写入端幂等重复无害。5. 常见问题与排查技巧实录5.1 报错 The $changeStream stage is only supported on replica sets这是新手最常撞的墙。原因就一个源库不是副本集模式。解决办法就是前面说的改配置加replSetName重启后rs.initiate()。注意rs.initiate()只需要执行一次重复执行会报 already initialized。如果你确认配置改了、也 initiate 了还是报这个错检查一下连接串里有没有带?replicaSetrs0。有些驱动在没带这个参数时会以直连模式连接直连模式下 Change Stream 是不工作的。这个坑很隐蔽因为连接本身是成功的只是 watch 会失败。5.2 同步延迟越来越大延迟增长通常有三个原因。一是写入端成了瓶颈如果目标库磁盘慢或者每条事件都同步写一次积压就会越来越多。解决办法是改成批量写攒够 100 条或 1 秒就 flush 一次。二是事件处理逻辑太重比如在转换层做了复杂的字段映射或远程调用拖慢了消费速度。三是网络带宽不够跨机房同步时如果 oplog 产生速度超过网络传输速度延迟必然增长。排查方法很直接在消费循环里打点记录事件产生时间和事件处理完成时间的差值。如果这个差值持续增长就是消费跟不上生产。先看是不是写入端慢再看是不是转换逻辑重最后才怀疑网络。5.3 目标库出现重复数据理论上幂等写入不该有重复但如果你的_id生成策略有问题就会出岔子。比如源库某些文档的_id是业务自己生成的字符串而目标库在写入时又用了insertOne而不是replaceOne那重复消费时就会插入两条_id相同但实际重复的文档——等等_id相同是插不进去的会报 duplicate key。所以如果你看到的是报错而不是重复数据那多半是没用 upsert。真正会出现重复数据的情况是源库的_id在目标库被重新生成了。比如转换层里手贱写了doc.pop(_id)让目标库自动生成新_id那每次重放都会生成一条新文档。这个错误很低级但很致命转换层里绝对不要动_id。5.4 常见问题速查表现象可能原因排查方向watch 报 replica set 错误源库非副本集 / 连接串缺 replicaSet检查配置和连接串同步延迟持续增长消费慢于生产打点测处理耗时改批量写目标库报 duplicate key未用 upsert改用 replaceOne upsert程序重启后漏数据token 未持久化或存错位置检查 token 存储逻辑delete 事件没同步pipeline 过滤掉了 delete检查$match条件update 事件 fullDocument 为空未设 updateLookup加full_documentupdateLookup跨机连不上主节点副本集 host 填了 127.0.0.1改成真实可达地址5.5 几个我踩过的坑坑一oplog 太小导致 resume 失败。如果同步程序停了很久而这段时间源库写入量很大oplog 可能已经把断点位置滚过去了。这时候用旧 token 恢复会报错。解决办法是把 oplog 调大--oplogSizeMB参数并且监控同步程序的存活状态别让它停太久。真滚过去了就只能重新全量。坑二大事务导致事件超大。如果源库有跨多文档的大事务Change Stream 会把整个事务作为一个事件推过来fullDocument可能非常大甚至超过 16MB 的文档限制。这种情况需要特殊处理比如把大事务拆开或者在转换层做流式处理。日常业务里少见但批量导入场景要留意。坑三时区问题。Change Stream 事件里的clusterTime是 UTC 时间戳如果你在转换层里拿它和本地时间比较很容易算错。统一用 UTC 处理展示时再转本地时区能避免很多莫名其妙的时间对不上问题。6. 性能调优与扩展思路6.1 批量写入把吞吐拉起来单条bulk_write其实和单条insertOne性能差不多真正的提升来自攒批。我的做法是维护一个缓冲区每收到一条事件就 append 进去当缓冲区达到 200 条或者距离上次 flush 超过 500 毫秒就一次性bulk_write出去。实测下来批量写入比单条写入吞吐能提升 5 到 10 倍尤其在跨机房高延迟场景下效果更明显。攒批的代价是延迟略微增加最多 500 毫秒以及崩溃时缓冲区里的数据会丢。对于延迟500 毫秒在大多数同步场景里可以接受。对于丢数据因为写入端幂等重启后从上次保存的 token 重放即可缓冲区里没写进去的事件会被重新消费。所以攒批和幂等是绝配缺一不可。6.2 多集合同步的并行化如果源库有多个集合要同步串行处理会互相拖累。可以给每个集合开一个独立的 Change Stream 和独立的消费线程各自维护自己的 token。这样某个集合的写入压力大不会影响其他集合的同步进度。不过并行度不是越高越好。每个 Change Stream 都会占用一个源库连接并且各自读 oplog。集合数量多的时候建议按业务相关性分组比如把同一业务模块的几个集合放一个线程里而不是一个集合一个线程。这样既控制了连接数又保证了业务内的顺序性。6.3 从单实例同步到 CDC 平台这套单实例同步方案其实是 CDC 的简化版。当你把监听、转换、写入三段跑通之后往平台化方向扩展就很自然了把配置抽出来做成元数据把 token 管理集中化把监控指标延迟、吞吐、错误率暴露出来再加一个 Web 界面管理同步任务。很多开源的 CDC 工具核心逻辑和这个骨架是相通的。我个人在实际操作中的体会是先把单集合、单方向的同步跑稳再考虑多集合、双向、平台化。一上来就追求大而全往往连最基本的不丢不重都保证不了。同步这件事可靠性永远排在功能丰富度前面。最后再分享一个小技巧调试阶段可以在转换层加一个影子模式只打印将要执行的操作而不真正写入目标库。这样能在不影响目标数据的前提下验证事件解析和转换逻辑是否正确。等确认无误了再打开真实写入。这个习惯帮我避免过好几次把测试数据写进生产库的事故。

单实例MongoDB数据同步：基于Change Stream的增量同步方案

相关推荐

AI Agent缓存系统设计：语义感知+三层架构实战

三车避障下基于TTC与安全距离的Carsim联合仿真轨迹规划

蓝牙音箱关机卡顿排查：从时序原理到杰理方案实操

更多资讯

人机协同：用 interrupt 把控制权交给你

Checkpointer 基础：把图的状态存下来

多轮对话：用持久化记住历史

收购通稿刷屏时，开发者社区在问什么？从 linux.do 帖子流与门户聚合页看 AI 信息流的供需错位

在 Python Web 开发生态中，Flask 始终占据着独特而重要的位置

深度学习1:基本概念，感知机，激活函数(阶跃，sigmoid,Tanh函数)

今日更新

电商智能运营体系从0到1：架构设计、落地步骤与避坑指南

压缩空气储能与热网热惯性联合优化调度：热电联产系统建模与求解

若依框架AI代码生成不稳定？用Agent Skill三层结构+五步法搞定

本周精选

金融AI落地指南：从调研报告到技术选型与工程实践

大模型上下文成本与注意力优化技术深度解析——为何O(n²)平方复杂度无法被常规优化消除｜TaoToken 统一 Key 通道实测

基于GMM的语音识别系统原理与MATLAB实现详解

本月热门

我发现了一个新思路：用 Remotion + Claude Code 像写代码一样自动化生成短视频

Windows下 Codex 中 Chrome 和 Computer Use 插件不可用问题排查及解决参考方式：TaoToken 统一 Key 配置与验证

2026 大模型集体涨价：用 Python 做企业 Token 成本测算与选型避坑（附配置）

读完文章，想聊聊您的网站？