MongoDB Change Streams 实时数据变更流处理实战指南

业务场景描述

在大型电商平台或高并发的在线系统中，业务数据的变更（如订单状态、库存变动、用户行为日志）需要实时通知下游系统，以便做流式分析、缓存更新或消息推送。传统的轮询方式不仅带来性能开销，还存在延迟较高的问题；而 Change Streams 能够基于 MongoDB 的副本集或分片集群，实现对集合、数据库乃至整个部署的实时数据变更订阅。

本文将结合真实生产环境场景，分享在微服务架构中，如何基于 MongoDB Change Streams 构建稳定、可扩展的实时变更流处理系统，并重点探讨遇到的坑及解决方案。

技术选型过程

目标需求
- 实时捕获指定集合或数据库中增删改数据，并可靠地推送给下游消费者。
- 支持消费端位点管理，以便应用重启或消费失败后能够继续消费。
- 可水平扩展，满足百万级写入的高吞吐量场景。
备选方案
- 轮询 Oplog：直接读取 MongoDB 的 oplog.rs 集合，进行数据解析推送。
- 使用 Kafka Connector：通过 Debezium 或 MongoDB 官方 Connector 将变更写入 Kafka。
- 原生 Change Streams：MongoDB 4.0+ 引入的标准化订阅接口，底层由副本集 oplog 驱动，不依赖第三方组件。
对比与决策
- 轮询 oplog 需要自行维护解析逻辑，耗时耗力且兼容性差。
- Kafka Connector 虽然成熟，但引入 Debezium 增加系统复杂度，并且 Connector 在分片集群上表现不够稳定。
- Change Streams 为官方一等公民，支持平滑横向扩展、位点存储灵活，且 API 简单易用。

最终决定使用原生 MongoDB Change Streams 方案。

实现方案详解

架构示意

┌──────────────┐       Change Streams        ┌───────────────┐
│  MongoDB 主副本集 │─────────────────────▶│  在线微服务消费 │
└──────────────┘                         └───────────────┘││▼┌────────────────┐│ 下游消息队列 (Kafka) │└────────────────┘

微服务 A 通过官方 MongoDB 驱动在启动时打开 Change Stream：
- 指定集合或数据库级别监控；
- 设置 fullDocument 选项以获取更新后的完整文档；
- 位点管理通过记录 resumeToken 实现。
实时消费变更事件后，将事件序列化并推送到 Kafka，供下游分析、缓存更新或异步通知使用。

Java Spring Boot 示例

// pom.xml 依赖
<dependency><groupId>org.mongodb</groupId><artifactId>mongodb-driver-sync</artifactId><version>4.5.0</version>
</dependency>// ChangeStreamListener.java
@Service
public class ChangeStreamListener {private final MongoClient mongoClient;private final KafkaTemplate<String, String> kafkaTemplate;private volatile BsonDocument resumeToken;public ChangeStreamListener(MongoClient mongoClient,KafkaTemplate<String, String> kafkaTemplate) {this.mongoClient = mongoClient;this.kafkaTemplate = kafkaTemplate;}@PostConstructpublic void startListening() {MongoDatabase db = mongoClient.getDatabase("orders_db");MongoCollection<Document> coll = db.getCollection("orders");ChangeStreamIterable<Document> stream = coll.watch().fullDocument(FullDocument.UPDATE_LOOKUP).resumeAfter(resumeToken);stream.forEach(change -> {// 保存位点resumeToken = change.getResumeToken();// 构建消息Document doc = change.getFullDocument();Map<String, Object> payload = new HashMap<>();payload.put("operationType", change.getOperationType().getValue());payload.put("data", doc);// 发送到 KafkakafkaTemplate.send("orders-change-topic", JSON.toJSONString(payload));});}
}

Node.js 示例

// 依赖: npm install mongodb kafkajs
const { MongoClient } = require('mongodb');
const { Kafka } = require('kafkajs');async function main() {const client = new MongoClient('mongodb://user:pwd@host:27017/?replicaSet=rs0');await client.connect();const kafka = new Kafka({ clientId: 'mongo-cs', brokers: ['kafka1:9092'] });const producer = kafka.producer();await producer.connect();const collection = client.db('orders_db').collection('orders');const changeStream = collection.watch([], { fullDocument: 'updateLookup' });changeStream.on('change', async (change) => {// 发送到 Kafkaconst message = {type: change.operationType,doc: change.fullDocument};await producer.send({topic: 'orders-change-topic',messages: [{ key: change._id.toString(), value: JSON.stringify(message) }]});});
}main().catch(console.error);

配置与部署

MongoDB 副本集开启 featureCompatibilityVersion 至 4.2+；
确保 maxAwaitTimeMS、batchSize 等参数根据业务量进行调整；
位点持久化可写入 Redis 或关系库，防止内存丢失导致消费重复或漏消费；
在 Kubernetes 中可部署多个副本消费实例，通过 resumeAfter 机制均衡分布负载。

踩过的坑与解决方案

Resume Token 过期
- 问题：使用长时间未消费导致 ResumeToken 过期，抛出 ChangeStreamNotFound 错误。
- 解决：捕获异常后，fallback 到最新游标（watch() 不带 resumeAfter）或从业务侧记录的时间点重新拉取变更。
网络抖动导致连接断裂
- 问题：短暂网络抖动导致 Change Stream 中断，消费逻辑重连时不知如何定位。
- 解决：在 finally 或 onError 中统一捕获断开事件，重试时使用上次保存的 resumeToken 进行恢复。
批量写入事件“丢失”
- 问题：大量插入场景下，默认 batchSize 导致事件被拆分，多次轮询才能完成一次批量写入，导致延迟。
- 解决：适当增大 batchSize、降低 maxAwaitTimeMS，并在消费端做合并或幂等处理。
下游消费端瓶颈
- 问题：推送到 Kafka 后，下游分析服务性能不足，导致 Topic 堆积。
- 解决：对高并发事件进行分区，使用多实例消费；或者在 Change Stream 消费层先进行汇总、限流处理。