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

1. 为什么单实例 MongoDB 之间也需要数据同步

很多人一听到"数据同步"，第一反应就是副本集、分片集群那一套。但实际工作中，我们遇到的场景往往没那么"高大上"——两台独立的服务器，各自跑着一个单实例 MongoDB，现在需要把 A 库里的部分数据搬到 B 库，或者让两边保持某种程度的同步。这种需求在数据迁移、灰度环境搭建、跨机房数据归集、测试环境数据回灌等场景里非常常见。

我最早接触这类需求是在一个数据归集项目里：边缘节点上跑着单实例 MongoDB，用来暂存采集到的数据，中心节点也是单实例，需要定期把边缘节点的数据汇总上来。当时第一反应是写脚本导出再导入，但数据量一上来，mongodump+mongorestore那套就顶不住了，而且没法做增量。后来才认真研究了 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()。这样这个实例就变成了一个"单节点副本集"，它会维护自己的 oplog，Change 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:pwd@source_host:27017/?replicaSet=rs0") db = client["your_source_db"] collection = db["your_collection"] pipeline = [ {"$match": {"operationType": {"$in": ["insert", "update", "replace", "delete"]}}} ] stream = collection.watch(pipeline, full_document="updateLookup")

几个关键点值得展开。pipeline参数允许你在服务端就过滤掉不关心的事件，比如只同步某几种操作类型，或者只同步某个字段满足条件的文档。服务端过滤能大幅减少网络传输，比拉回来再过滤高效得多。

full_document="updateLookup"这个参数非常重要。默认情况下，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 走 upsert，delete 走 deleteOne。

有个细节要注意：update 事件的fullDocument只有在设了updateLookup时才有，而且它反映的是"事件被处理那一刻"的文档状态，不是"变更发生那一刻"的状态。如果同一个文档在极短时间内被改了多次，你拿到的可能是最新版本。这在绝大多数场景下没问题，但如果你的业务对中间状态敏感，就得改用full_document_before_change（MongoDB 6.0+ 支持）来拿变更前的快照。

4. 实操过程与核心环节实现

4.1 环境准备清单

动手前先把这些东西备齐，能省掉大量来回折腾的时间：

版本这块我建议源库至少 4.4。3.6 虽然支持 Change Stream，但早期版本在 resume token 的稳定性和updateLookup的性能上都有欠缺，4.4 之后成熟很多。如果条件允许，直接上 5.0 或 6.0，full_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, upsert=True) 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], ordered=False) self.save_token(event["_id"]) except PyMongoError as e: print(f"stream error: {e}, retrying in 3s") time.sleep(3) if __name__ == "__main__": syncer = MongoSyncer( source_uri="mongodb://sync_reader:pwd@source_host:27017/?replicaSet=rs0", target_uri="mongodb://writer:pwd@target_host:27017/", db_name="your_source_db", coll_name="your_collection", token_file="./resume_token.txt" ) syncer.run()

这段代码有几个设计取舍值得说。用bulk_write而不是单条写，是为了将来扩展成批量处理时不用改结构，现在一次写一条也没问题。ordered=False表示不要求顺序执行，遇到单条失败不影响其他条，配合幂等写入很合适。外层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()}}, upsert=True )

这样 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 了还是报这个错，检查一下连接串里有没有带?replicaSet=rs0。有些驱动在没带这个参数时会以直连模式连接，直连模式下 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 常见问题速查表

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 工具，核心逻辑和这个骨架是相通的。

我个人在实际操作中的体会是，先把单集合、单方向的同步跑稳，再考虑多集合、双向、平台化。一上来就追求大而全，往往连最基本的"不丢不重"都保证不了。同步这件事，可靠性永远排在功能丰富度前面。

最后再分享一个小技巧：调试阶段可以在转换层加一个"影子模式"，只打印将要执行的操作而不真正写入目标库。这样能在不影响目标数据的前提下，验证事件解析和转换逻辑是否正确。等确认无误了，再打开真实写入。这个习惯帮我避免过好几次"把测试数据写进生产库"的事故。