1. 项目到底在同步什么我为什么会对MongoDB产生“纠缠”的念头做跨机房服务的时候我接了一个很拧巴的需求两套MongoDB副本集分布在不同的城市各自都有写入流量业务方却要求它们的数据在几秒内保持一致而且不能出现明显的“主人”角色——任何一端写入另一端都必须能感知到并自动补齐。跑了一段时间后我突然觉得这不就是“量子纠缠”吗A节点发生变化B节点几乎同时就能“感应”到并做出反应完全不依赖于轮询式地反复去问“你有没有新数据”。单从名词上讲这种同步更像是一种高可用的数据传播语义。它要求的不只是把数据从A复制到B而是让两个库之间的状态持续收敛中间还包括了全量初始化、增量变更捕获、冲突合并、断点续传、数据校验这些环节。对于.NET开发者来说MongoDB.Driver并不难上手难的是把“同步”这件事做到业务可接受的程度延迟不能超过5秒不能丢数据不能因为网络抖动就全线崩掉写冲突的时候还要保证最后的状态是合理的。我当时把方案拆成了三块第一块是数据变更的实时感知第二块是变更事件在.NET进程里的可靠传递第三块是目标端的幂等写人与冲突修正。拆完发现市面上大部分同步工具解决的是前两块但真正花时间的往往是第三块。而且MongoDB的Change Streams、游标断点、集群状态切换这些概念官方文档里每个词都认识串在一起却不告诉你该怎么落地。这篇文章就把我踩过的坑和最终跑通的方案拆开讲适合正在做多机房容灾、业务双活或者只是想把两个环境的MongoDB数据保持一致的人参考。如果你也遇到类似问题先别急着去搭Kafka或者上同步中间件。很多场景下MongoDB自带的Change Streams配合.NET的Channel就足够了。关键在于怎么设计检查点、怎么处理重复投递、怎么在双向写入时不让数据“打架”。我把这些核心经验全部记下来按从简到繁的顺序展示。1.1 业务场景还原所谓“跨时空”到底是什么项目背景是一个两中心双活场景。中心A部署了订单服务和对应的MongoDB副本集中心B也部署了一套相同结构的应用和MongoDB。正常情况下用户流量按地域路由但两组MongoDB都允许写入也就是说用户可能在北京改了自己的一条紧急订单同时在上海的另一个运营后台也改了同一条订单的备注。这种架构下单靠MongoDB原生的副本集是解决不了问题的因为原生复制是主从复制只能有一个primary写入点跨机房网络抖动时还会导致主节点频繁切换甚至脑裂。所以项目要做的其实是业务层面的双向同步。它有两个明显特征一是数据源不唯一任何一端都可能产生修改二是延迟要求高运营后台要能在秒级看到另一端的修改结果。如果只是做主备简单单向同步就够了但双活必须把“同步”从备份提升为“状态融合”。我会在后面的小节里讲清楚为什么不用原生复制集以及自己动手实现同步时最容易犯的致命错误。1.2 为什么“复制集同步”解决不了双活问题先澄清一点MongoDB副本集本身就有非常完善的复制机制。主节点写完后oplog会给从节点同步从节点异步应用这些操作。如果只是需要一个从库来读取或备份完全不用自研同步方案。但双活场景下两个副本集都需要独立对外提供写服务你不可能让两个副本集的primary同时存在还自动合并逻辑冲突。另一个隐患是网络故障下的主从切换。跨机房网络一旦抖动副本集内部的心跳检测就会触发重新选举通常30秒左右才能恢复写入。业务对这种级别的间断是无法容忍的毕竟用户只知道“点了一下保存怎么没反应”。而如果自己实现业务层同步可以在网络恢复后做增量补偿给业务平滑过渡的机会。再一个原因原生复制的单元是oplog条目两个副本集之间的层级关系是定死的一旦断链只能重新全量同步。真要做双活就得把冲突处理规则掌握在自己手里比如“最后写入覆盖”或者“字段级别合并”这些都是原生机制给不了的。所以当时我决定在.NET服务里嵌入一套同步引擎用MongoDB驱动监听一个库的变更再把变更应用到另一个库。1.3 “量子纠缠”式同步的精确定义我说“量子纠缠”其实是在描述一种行为特征任何一个库里的数据变化会被另一个库实时感知并自动收敛。这种同步不是简单的复制粘贴而是由事件触发、带状态跟踪、循环往复的传播过程。拆开来看包含三个核心能力实时感知说的是能捕获到最细粒度的数据变化自动传播说的是变化能及时投递到目标端自动恢复说的是断流后能从断点继续处理不漏不重。实际做的时候实时感知靠的是Change Streams自动传播靠的是事件队列自动恢复靠的是持久化游标检查点。听起来简单但每一步都有不少细节。我见过很多人用轮询方式做“伪实时”同步定时查表中的更新时间字段这样实现简单但删除记录时抓不到而且延迟无法压缩到秒级。用Change Streams之后同步延迟可以压到毫秒级前提是正确处理好事件顺序和分批应用。2. 选型之前的硬核思考轮询、Change Streams还是消息中间件既然要自己实现同步那你一定会面对三个技术方向定期轮询、Change Streams事件监听、接入外部消息队列。我全部在项目里试过直接说结论低延迟双活推荐Change Streams为主体消息队列作为缓冲轮询只用于全量对账和补偿。不过要理解为什么得从每个方案的底层机制讲起。2.1 轮询同步的问题延迟和删除盲区最直观的同步方案是定期查询“最近更新时间大于上次游标”的文档然后把它们搬到目标库。我最初优化就是先上轮询10秒跑一次发现业务根本接受不了因为运营后台里一个备注修改要等10秒才能看到体验非常糟糕。把轮询周期缩短到1秒以后MongoDB压力明显上来因为每次都要对一个大集合做带排序的查询即使有索引海量集合下成本也不低。更致命的问题是轮询无法捕获删除操作。如果一条记录被物理删除你在源库查不到它自然不知道要删掉目标库里对应的记录。有人靠软删除字段规避但强制业务侧加字段总是不太靠谱。对比之下Change Streams直接把insert、update、delete、replace事件暴露给你删除处理就顺理成章了。所以轮询只适合做兜底的全量校验不适合做增量主链路。2.2 Change StreamsMongoDB自带的“事件雷达”MongoDB 3.6开始提供Change Streams核心原理是监听oplog然后以事件流的形式把变更推送出来。它可以监听整个数据库、某个集合甚至聚合管道过滤后的数据。和轮询相比它的优势是实时修改一落库事件几乎同时到达客户端还附带着完整文档的旧值和新值通过pre/postImages配置。在.NET里使用Change Streams不需要额外安装什么组件官方驱动MongoDB.Driver里已经有WatchAsync方法。你把游标拿到用foreach循环等待即可。这里的“游标”不是普通查询游标它会一直挂在那里有新数据就返回没数据就阻塞。正因如此Change Streams天然适合做“订阅-推送”模式不需要你反复发查询请求。需要提醒的是Change Streams要求MongoDB必须以副本集或分片集群方式部署独立版本不支持。如果你本地只有一个mongod进程要先把节点初始化成单节点副本集。这一步经常有人忘记后面我会给出具体命令。2.3 为什么选择.NET的Channel做事件缓冲Change Streams的事件到达客户端之后如果直接在遍历循环里写目标库会因为网络IO阻塞住整个流。假设同步进程每分钟要处理上千条变更目标库偶尔慢一下你的监听循环就卡住了后续变更无法及时消费。所以我引入了System.Threading.Channels用无界Channel作为事件缓冲。Channel是.NET里一个异步生产者-消费者队列比手写Queue加锁要优雅得多。生产者把ChangeStream事件写入Channel消费者从Channel里读取并批量应用到目标库。这样即使目标库慢监听端也能快速把事件收下来放进内存不会阻塞MongoDB的游标读取。选择Channel而不是Kafka或RabbitMQ主要是为了减少运维成本。项目规模不大十几台机器用消息中间件反而是负担。但如果你有多套目标库、多种消费者、需要长时间回放事件那还是上Kafka更稳。我们当时的逻辑是先用Channel跑通等真需要多消费者时再平滑迁移到消息队列。这个思路值得参考不要一上来就为未来三个月的需求过度设计。2.4 双向同步的困惑谁才是“主纠缠态”双活意味着两个库都能写但同步引擎在把A的变更同步到B的时候如果B也在写相同的记录就会发生冲突。我一开始天真地认为只要把两边事件都互相传播就行结果立刻陷入循环A的写入同步到BB因为那次写入又生成新变更又同步回AA又产生新事件……这成了无限套娃数据也是一遍一遍地被搬来搬去。归根到底要定一个“权威问题”的答案。通常有两种策略一种是在业务规则上为不同对象指定不同的写入中心比如订单在A中心写入商品配置在B中心写入另一种是给每条记录加版本号或时间戳在冲突时按照最后写入覆盖或字段级合并。第二种更像是通用方案我会在第四节详细解读。不过你要明白双向同步并不是简单地把两个单向同步叠加。背后必须有稳定的去重机制防止自己在给自己“回声”。去重最简单的方式是在事件里带上来源标识处理时忽略本端自己产生的事件。如果再配合版本号合并基本就能避免大部分冲突。3. 动手之前的准备MongoDB环境、.NET驱动与最小监听示例不知道你是从哪个阶段开始搜到这篇文章的。很多人问我“我MongoDB都还没安装这部分会不会跳过”。所以我特意把环境准备写全从零开始带你搭建。已经装过MongoDB的人可以直接跳到3.3节但建议还是看一眼副本集初始化部分那是不开Change Streams最常见的坑。3.1 MongoDB安装与副本集初始化Windows环境下直接去MongoDB官网下载MSI安装包一路Next即可。安装过程中最好选择“Install MongoDB as a Service”和“Install MongoDB Compass”前者让你无需手动启动服务后者方便可视化查看同步结果。Linux环境下比如Debian系需要先把官方仓库导入apt源再用apt install mongodb-org安装安装后用一个配置文件启动。启动后第一件事不是连接而是把MongoDB初始化为单节点副本集。Change Streams不支持单机模式所以需要往mongod.conf里加一行replication.replSetName比如设置为rs0然后重启服务。接着进入mongosh执行rs.initiate()这一句就能把一个普通节点变成副本集primary。执行完再用rs.status()查看状态。如果显示stateStr: PRIMARY说明副本集已经跑起来了。很多人忽略这一步后面在.NET里调用Watch时会直接报“The $changeStream stage is only supported on replica sets”而且这个错误还特别容易误以为是驱动版本问题。3.2 在.NET项目里引入MongoDB.Driver并配置连接创建一个控制台应用程序或者用ASP.NET Core的BackgroundService都行。我习惯用.NET 8包管理里直接搜索MongoDB.Driver版本至少2.19以上因为新版驱动对Change Streams和序列化的支持更完善。下面是安装命令dotnet add package MongoDB.Driver连接字符串需要指向副本集的名字比如mongodb://localhost:27017/?replicaSetrs0。如果你有多个节点把它们都写上驱动会自动做节点发现和探活。然后创建一个MongoClient实例拿到数据库和集合var client new MongoClient(mongodb://localhost:27017/?replicaSetrs0); var database client.GetDatabase(orders); var collection database.GetCollectionBsonDocument(order);这里我把集合定义为BsonDocument因为同步引擎要处理各种不同结构的文档用强类型反而限制住。如果你同步的是特定实体可以定义POCO类然后配置映射但通用场景里BsonDocument更灵活。3.3 第一个能跑的变更监听器现在到最激动人心的环节用.NET监听MongoDB集合的每一次变化。在MongoDB.Driver里调用Watch方法会返回一个IChangeStreamCursor你可以用MoveNextAsync或ForEachAsync来循环接收事件。最简单的一段代码var options new ChangeStreamOptions { FullDocument ChangeStreamFullDocumentOption.UpdateLookup }; using var cursor await collection.WatchAsync(options); await cursor.ForEachAsync(change { Console.WriteLine($操作类型: {change.OperationType}); Console.WriteLine($文档Id: {change.FullDocument?[_id]}); });这段代码会让控制台一直挂着每当order集合插入或更新一条文档都会打印出来。FullDocument UpdateLookup表示在update事件里通过FullDocument字段附带修改后的最新文档省得你根据文档ID再去查一次。如果是第一次写你可能会发现它确实能捕获到其他客户端通过mongosh或Compass执行的修改非常灵敏。但这里有一个陷阱如果你用同一个C#程序的InsertOne也能触发事件吗能因为事件是MongoDB服务端发出的和客户端无关。不过即便如此我们的同步引擎还是要做好来源标记否则会出现上一节提到的“无限套娃”。3.4 用Channel实现事件缓冲与批量落库单独监听很好玩但直接应用到目标库前最好先把事件排队。我实现了一个“生产者-消费者”模式生产者是Watch循环消费者是后台任务。在.NET里Channel的用法非常简单var channel Channel.CreateUnboundedChangeStreamDocumentBsonDocument(); var producer Task.Run(async () { using var cursor await collection.WatchAsync(options); await foreach (var change in cursor.ToEnumerable()) { await channel.Writer.WriteAsync(change); } }); var consumer Task.Run(async () { await foreach (var change in channel.Reader.ReadAllAsync()) { // 这里把事件应用到目标数据库 } });这样做的好处是生产端和消费端解耦。生产端永远只是把事件塞进Channel即使目标库写得很慢源库的游标也能一直被消费不存在因为下游阻塞而上游迟迟不读事件的情况。当然无界Channel有内存溢出风险生产环境一般用有界Channel并设置BoundedCapacity再把WaitToWriteAsync的异步等待做好。这套模式是我觉得比较稳妥的中间过渡方案。4. 完整版实现高可靠的增量同步、断点续传与冲突消解当你把最小监听跑通之后就会面临三个逼疯人的问题进程重启怎么办、重复事件怎么办、两边同时写怎么办。这一部分就是整个同步方案的核心我不绕弯子直接给出能用于生产的代码思路和配置细节。4.1 持久化检查点用resume token接续“纠缠”Change Streams支持通过resumeToken从某个历史位置恢复监听这个令牌是一个唯一的时间位置标识。比如进程处理到第N个事件你把该事件的ResumeToken保存到MongoDB的一个专用集合里重启后用它作为ResumeAfter重新建立Change Stream就能接着处理而不必重新全量同步。保存令牌的代码如下var resumeToken change.ResumeToken; await checkpointCollection.UpdateOneAsync( BuildersBsonDocument.Filter.Eq(_id, sync_checkpoint), BuildersBsonDocument.Set.Set(token, resumeToken), new UpdateOptions { IsUpsert true } );每次处理完一批事件后再更新令牌而不是每收到一个事件就更新目的是降低写放大。但是要注意如果进程在批处理中途崩溃已经处理完的事件还没来得及更新令牌重启后这些事件会被重复消费所以目标端必须保持幂等。在ChangeStreamOptions里设置ResumeAfter即可恢复var resumeToken GetSavedResumeToken(); var options new ChangeStreamOptions { FullDocument ChangeStreamFullDocumentOption.UpdateLookup, ResumeAfter resumeToken };如果ResumeAfter指定的令牌太旧MongoDB会因为oplog被覆盖而抛出ChangeStreamHistoryLost异常。这时需要退回到全量同步。处理这个异常的代码必须写在业务里而不是只打印日志。我通常的做法是捕获到MongoCommandException且错误码为286或280时重新执行全量拷贝并清空保存的checkpoint。4.2 幂等写入让每次同步都“重新执行也不怕”重复消费并不可怕可怕的是重复消费后目标库数据被弄坏。MongoDB的ReplaceOne加上_id过滤天然幂等如果目标库已经有这条记录就覆盖没有就插入。对于insert和replace事件我直接var filter BuildersBsonDocument.Filter.Eq(_id, change.FullDocument[_id]); await targetCollection.ReplaceOneAsync( filter, change.FullDocument, new ReplaceOptions { IsUpsert true } );对于delete事件则按_id删除await targetCollection.DeleteOneAsync(filter);为什么不用InsertOne因为update事件可能先一步到达目标库已经有这条记录了再调InsertOne会报主键冲突。用ReplaceOne加upsert无论事件顺序怎么乱都能保证最终一致。这是很多自己写同步的人容易忽略的地方。4.3 双向同步的冲突消解版本号与最后写入双向同步最烦的就是两边同时改同一条记录。我的方案是给集合加一个sync_version字段每次业务写入时更新这个版本号。同步引擎在应用目标对象前会比较本地版本号和事件中的版本号如果事件版本号大于本地版本号就执行写入否则跳过。还有一种常见策略是Last-Write-Wins按时间戳决定。但时间在不同服务器上会有偏移用版本号或单调递增计数器更可靠。具体做法是var eventVersion change.FullDocument.GetValue(sync_version, 0).AsInt32; var localDoc await targetCollection.Find(filter).FirstOrDefaultAsync(); var localVersion localDoc?.GetValue(sync_version, 0).AsInt32 ?? -1; if (eventVersion localVersion) { await targetCollection.ReplaceOneAsync(filter, change.FullDocument, new ReplaceOptions { IsUpsert true }); }如果业务无法增加字段也可以把文档的其他参与业务判断的关键字段取出来做哈希但可维护性不如版本号方案。从我实测下来的情况看版本号在绝大多数业务逻辑里都能接受代价只是一个整数索引。4.4 全量初始化与对账补偿机制增量同步只能解决从当前时间点之后的变化对于变更流历史丢失、或者源库在很久之前就有大量数据的情况必须先做一次全量初始化。全量初始化不能直接在监听线程里做因为数据量大的时候会阻塞增量事件。我习惯先关掉Change Stream消费者然后把源集合的数据分批拷贝到目标集合拷贝完成后记录一个initial_sync_time再从这个时间点开始重新开启监听并对可能发生的重复进行幂等处理。但这个方案依然有缝隙全量拷贝过程中业务数据还在变化如果监听从拷贝完成后开始中间修改的数据就丢了。所以严格一点的做法是在全量开始前记录一个startAtOperationTime全量拷贝完成后用这个时间点开启Change Stream。MongoDB 4.0以上支持通过StartAtOperationTime指定时间点恢复比用resume token更简洁。除了初始化定时对账也很有必要。我写了一个每小时跑一次的任务抽样对比两个库的文档数和几个关键字段的哈希值发现不一致就触发按源库全量覆盖。对账不能太频繁否则会拖垮性能。5. 实战踩坑记录从连接卡死到“纠缠态”退相干如果一切按部就班运行你怀疑自己是不是只写了个Hello World。真实的生产环境从来不会安静下面几类问题我全遇到过而且非常典型。5.1 net::ERR_INCOMPLETE_CHUNKED_ENCODING响应体被截断同步服务提供了一个HTTP接口用来给运营后台展示当前同步进度和健康状态。某次部署后浏览器里刷新后台发现报错net::ERR_INCOMPLETE_CHUNKED_ENCODING提示响应不完整。但我看服务日志一切正常接口返回的数据也不算大。排查后发现这不是MongoDB同步本身的问题而是HTTP层有一个反向代理在传输大响应时等不到结尾连接被重置。当时同步进度接口一次性返回了上万条队列积压明细响应用了chunked编码代理服务器的超时时间设置太短直接掐断了连接。这件事给我的教训是同步服务对外暴露接口时要么对大数据量接口做分页要么直接关闭代理缓冲改成流式返回。如果是.NET的WebApplication可以给接口返回IAsyncEnumerable并设置不做缓冲响应如果必须返回普通JSON尽可能限制条数。遇到这个报错不要先怀疑MongoDB先看返回体大小和代理超时配置。5.2 Change Streams停止后丢了增量数据有一次机房网络闪断同步进程还活着但MongoDB连接被切断。重连后我的代码是直接WatchAsync没用检查点恢复导致断连期间的数据全部丢失。最让人崩溃的是Change Streams本身不会报错给你说“有数据丢了”它只会从当前时间开始继续发送。显然如果没有任何持久化检查点这种丢失是静默的。解决办法就是我在4.1里写的resume token持久化。另外网络重连后不能立刻从头打开游标要先尝试从上次保存的token恢复。如果oplog保留时间太短还需要把MongoDB的oplogSizeMB调大或者用时间窗口管理法。我自己的项目里把oplog保留时间设为3小时足够覆盖对账周期。5.3 权限配置不当导致Watch时报“not authorized”很多同步服务使用单独的MongoDB账号而不是root。Change Streams这个操作需要的权限是changeStream同时还要有对集合的find权限。如果只是给了readWrite运行一段时间后会发现监听一直有效直到节点重新选举或客户端重连才突然报not authorized非常迷惑。建议在MongoDB里为同步用户专门分配角色db.createUser({ user: syncUser, pwd: your_password, roles: [ { role: readWrite, db: orders }, { role: read, db: local } ] });其中local库的读权限是为了读取oplog这是Change Streams内部依赖的。如果你分片集群可能还需要更多权限。具体以官方文档为准但中心思想是账号权限不能只照搬普通应用账号。5.4 性能监控如何判断“纠缠”是否正在同步同步系统最怕的是表面风平浪静实际上早已断线。我在服务里加了一套可视化监控通过EventCounter统计每秒处理事件数、Channel积压数量、最近一次心跳时间。只要Channel积压持续上涨说明消费者性能跟不上如果最近一次心跳时间超过30秒说明Change Streams可能断了。另外MongoDB官方也提供db.currentOp()和$changeStream事件指标可以配合观察。不过对于.NET团队最直接的是把所有关键指标暴露给Prometheus配上Grafana看板。比起各种花哨日志实时曲线更能让你一眼看出同步是否处于“健康纠缠”状态。6. 一点经验分享如果让我重做我会怎么下手做这套同步方案前后花了将近三周其中一半时间都花在冲突处理和故障恢复上。如果让我重来一次我会先建一个最简单的单向同步跑通加上检查和断点续传再启动双向支持。不要一开始就想着支持字段级合并或者多租户隔离复杂设计会拖慢你的交付节奏。另外一个比较大的感触是事件驱动同步的核心不是“怎么监听”而是“怎么在故障后收敛”。监听事件只是基础能力真正决定系统可靠性的是你怎么处理重复、怎么恢复断点、怎么验证目标库和源库最终一致。与其精通各种花哨特性不如把幂等性做扎实。最后还有个小技巧可以分享在Change Streams里如果业务上允许高延迟可以批量更新多个文档而不是一条一条更新。很多变更事件都集中在一张表上把它们按_id分组后集合到同一个批次里批量WriteModel性能能提升一个数量级。我们线上最终就是按这个方式跑的同步延迟始终稳定在1秒以内日志里也再没有出现惊慌失措的告警。