Change Streams 是 MongoDB 3.6 引入的变更流功能，允许应用以拉取或推送的方式实时监听集合、数据库甚至整个集群的数据变更。它是构建事件驱动架构、数据同步、缓存失效、审计日志等场景的核心基础设施。Change Streams 底层基于 oplog（操作日志），天然继承了 MongoDB 复制机制的高可用与顺序性保证，同时对外屏蔽了 oplog 的复杂格式，提供结构化的变更事件流。与直接轮询相比，Change Streams 具有低延迟、低负载、强一致的优势，已成为 MongoDB 生态中实现 CDC（Change Data Capture）的首选方案。
1. 基本原理
1.1 Change Streams 与 oplog 的关系
MongoDB 的复制集通过 oplog 记录所有写操作，Secondary 节点异步拉取并回放 oplog 以保持数据一致。Change Streams 本质上是在 oplog 之上构建的「安全包装层」：MongoDB 服务器内部维护一个 Change Stream 游标，将 oplog 中的原始条目解析为结构化事件，再通过长连接（或轮询游标）推送给客户端。
┌─────────────┐ ┌──────────────┐ ┌──────────────────┐
│ 应用层 │←────│ Change Stream │←────│ oplog │
│ (Consumer) │ │ 游标/解析 │ │ (capped collection)│
└─────────────┘ └──────────────┘ └──────────────────┘
关键点：
- Change Streams 只能运行在复制集（Replica Set）或分片集群（Sharded Cluster）上，因为单机实例没有 oplog。
- 每个变更事件携带
clusterTime（逻辑时间戳），保证全局有序。 - Change Stream 游标最终读取的是主节点（Primary）的 oplog，因此网络分区或主节点切换可能导致短暂中断，需要配合 Resume Token 做容错恢复。
1.2 事件类型详解
| 事件类型 | 说明 | 适用场景 |
|---|---|---|
insert | 新文档插入 | 数据同步、缓存预热 |
update | 文档字段更新 | 增量索引、通知触发 |
replace | 整文档替换 | 全量覆盖场景 |
delete | 文档删除 | 缓存失效、级联删除 |
invalidate | 监听目标被删除或重命名 | 需要重新建立 Change Stream |
drop | 集合被删除 | 清理下游资源 |
1.3 事件结构字段完整说明
{
"_id": { "_data": "8262..." },
"operationType": "update",
"fullDocument": { "_id": "...", "status": "shipped", "updatedAt": "..." },
"ns": { "db": "shop", "coll": "orders" },
"documentKey": { "_id": "..." },
"updateDescription": {
"updatedFields": { "status": "shipped" },
"removedFields": [],
"truncatedArrays": []
},
"clusterTime": { "$timestamp": { "t": 1692000000, "i": 1 } },
"txnNumber": 1,
"lsid": { "id": "...", "uid": "..." }
}
_id._data：Resume Token，用于断点续传。operationType：事件类型。fullDocument：完整文档，只有insert、replace或配置了updateLookup的update才包含。ns：Namespace，标识数据库与集合。documentKey：被变更文档的_id。updateDescription：update事件中描述字段变更细节。clusterTime：事件在集群中的逻辑时间戳，用于排序和一致性判断。txnNumber/lsid：多文档事务相关标识。
2. 基础用法
2.1 Mongo Shell
const pipeline = [
{ $match: { "fullDocument.status": "paid" } },
{ $project: { "fullDocument": 1, "operationType": 1, "clusterTime": 1 } }
];
const changeStream = db.orders.watch(pipeline, {
fullDocument: "updateLookup"
});
changeStream.on("change", (change) => {
console.log(change);
});
2.2 Java 驱动
MongoClient client = MongoClients.create("mongodb://localhost:27017");
MongoDatabase db = client.getDatabase("shop");
MongoChangeStreamCursor<ChangeStreamDocument<Document>> cursor =
db.getCollection("orders").watch()
.fullDocument(FullDocument.UPDATE_LOOKUP)
.cursor();
while (cursor.hasNext()) {
ChangeStreamDocument<Document> change = cursor.next();
switch (change.getOperationType()) {
case INSERT -> handleInsert(change.getFullDocument());
case UPDATE -> handleUpdate(change.getFullDocument());
case DELETE -> handleDelete(change.getDocumentKey());
}
}
2.3 Node.js 驱动（mongodb 包）
const { MongoClient } = require('mongodb');
async function run() {
const client = new MongoClient('mongodb://localhost:27017');
await client.connect();
const db = client.db('shop');
const pipeline = [
{ $match: { operationType: { $in: ['insert', 'update'] } } },
{ $project: { fullDocument: 1, ns: 1, operationType: 1 } }
];
const changeStream = db.collection('orders').watch(pipeline, {
fullDocument: 'updateLookup',
maxAwaitTimeMS: 3000
});
for await (const change of changeStream) {
console.log('Event:', change.operationType, 'on', change.ns.coll);
if (change.fullDocument) {
console.log('Doc:', change.fullDocument);
}
}
}
run().catch(console.error);
2.4 Python（pymongo）驱动
from pymongo import MongoClient
client = MongoClient('mongodb://localhost:27017')
db = client['shop']
pipeline = [
{"$match": {"operationType": {"$in": ["insert", "update"]}}},
{"$project": {"fullDocument": 1, "ns": 1, "operationType": 1}}
]
with db.orders.watch(pipeline, full_document='updateLookup') as stream:
for change in stream:
print(f"Type: {change['operationType']}")
print(f"Token: {change['_id']}")
if 'fullDocument' in change:
print(f"Doc: {change['fullDocument']}")
2.5 过滤管道详解
Change Streams 支持完整的聚合管道阶段，常用阶段包括：
| 阶段 | 作用 | 示例 |
|---|---|---|
$match | 按条件过滤事件 | { $match: { "fullDocument.status": "paid" } } |
$project | 投影，减少字段 | { $project: { "fullDocument.name": 1, "operationType": 1 } } |
$addFields | 添加派生字段 | { $addFields: { eventTime: "$clusterTime" } } |
$redact | 基于条件控制字段可见性 | 可结合 $$DESCEND / $$PRUNE 做安全过滤 |
2.6 fullDocument 选项
| 选项值 | 说明 |
|---|---|
default | update 事件不包含完整文档 |
updateLookup | update 事件会查询当前文档状态并返回 |
required | 要求必须返回完整文档，否则报错 |
whenAvailable | 当 post-image 可用时返回（MongoDB 6.0+） |
MongoDB 6.0 开始支持 changeStreamPreAndPostImages，可以在集合级别启用前置/后置镜像，Change Stream 通过 fullDocumentBeforeChange 获取更新前的文档快照。
db.runCommand({
collMod: "orders",
changeStreamPreAndPostImages: { enabled: true }
});
3. Resume Token 与容错
3.1 Resume Token 序列化与反序列化
Resume Token 本质是 BSON 格式的 _id 字段，需要安全地序列化为字符串后持久化到外部存储（Redis、ZooKeeper、数据库等）。
// 序列化：Token → String
String tokenString = resumeToken.toJson();
redisTemplate.opsForValue().set("changeStream:orders:token", tokenString);
// 反序列化：String → BsonDocument
String saved = redisTemplate.opsForValue().get("changeStream:orders:token");
BsonDocument restoredToken = BsonDocument.parse(saved);
3.2 Token TTL 与 oplog 窗口关系
Resume Token 的有效期取决于 oplog 的保留窗口。如果 Token 指向的 oplog 条目已被覆盖（oplog 淘汰），MongoDB 将抛出 ChangeStreamFatalError，此时无法 Resume，只能从头重新监听。
因此生产环境中需监控 oplog.rs 的窗口大小：
db.getReplicationInfo().logLengthStartToEnd;
建议：
- 配置足够大的 oplog（根据写吞吐量评估）。
- 消费端发生故障时尽快恢复，避免 Token 过期。
- 定期将当前的 Resume Token 保存到可靠的外部存储。
3.3 批量处理与背压控制
在高吞吐场景下，逐条处理变更事件可能成为瓶颈。可以利用驱动提供的批量游标接口实现背压控制：
ChangeStreamIterable<Document> iterable = db.getCollection("events").watch()
.batchSize(100)
.maxAwaitTime(5, TimeUnit.SECONDS);
try (MongoChangeStreamCursor<ChangeStreamDocument<Document>> cursor = iterable.cursor()) {
List<ChangeStreamDocument<Document>> batch = new ArrayList<>();
while (cursor.hasNext()) {
batch.add(cursor.next());
if (batch.size() >= 100) {
processBatch(batch);
saveTokenToRedis(batch.get(batch.size() - 1).getResumeToken());
batch.clear();
}
}
}
3.4 断线重连指数退避策略
BsonDocument resumeToken = loadTokenFromRedis();
int retryCount = 0;
final int MAX_BACKOFF_MS = 60000;
while (true) {
try {
ChangeStreamIterable<Document> iterable;
if (resumeToken != null) {
iterable = db.getCollection("orders").watch().resumeAfter(resumeToken);
} else {
iterable = db.getCollection("orders").watch();
}
try (MongoChangeStreamCursor<ChangeStreamDocument<Document>> cursor = iterable.cursor()) {
retryCount = 0; // 连接成功后重置计数器
while (cursor.hasNext()) {
ChangeStreamDocument<Document> change = cursor.next();
process(change);
resumeToken = change.getResumeToken();
saveTokenToRedis(resumeToken);
}
}
} catch (MongoException e) {
int backoff = Math.min((int) Math.pow(2, retryCount) * 1000, MAX_BACKOFF_MS);
log.warn("Stream interrupted, retry {} after {} ms", retryCount, backoff);
retryCount++;
Thread.sleep(backoff);
}
}
4. 常见 CDC 架构
4.1 MongoDB → Debezium → Kafka Connect
Debezium 是开源的分布式 CDC 平台，内置 MongoDB Connector。它利用 Change Streams 捕获变更，将事件序列化为 Kafka 消息，供下游消费者订阅。
架构示意：
MongoDB Primary ──Change Streams──→ Debezium Connector
│
▼
Kafka Topic
│
┌─────────────────────┼─────────────────────┐
▼ ▼ ▼
ElasticSearch ClickHouse 业务服务
(全文检索) (OLAP 分析) (事件处理)
Debezium 配置要点：
capture.mode：设置为change_streams_update_full以获取完整文档。snapshot.mode：首次启动时可选择全量快照 + 增量流。tombstones.on.delete：控制是否对delete事件发送 tombstone 消息。
4.2 MongoDB → AWS Lambda 触发器
MongoDB Atlas 提供与 AWS 的原生集成，支持在集合上配置触发器（Trigger），当 Change Stream 产生事件时自动调用 AWS Lambda 函数。
// AWS Lambda 处理函数示例（Node.js）
exports.handler = async (event) => {
const changeEvent = event.fullDocument;
if (event.operationType === 'insert') {
await sendWelcomeEmail(changeEvent.email);
}
return { statusCode: 200 };
};
优势：无需自建长连接消费者，Serverless 弹性扩缩容。
劣势：事件延迟受 Atlas 触发器轮询间隔影响（通常数百毫秒到数秒）。
4.3 MongoDB → Elasticsearch 实时索引同步
通过 Change Streams 将 MongoDB 中的文档变更实时同步到 Elasticsearch，是最常见的搜索增强架构。
// 简化示例：update 事件同步到 ES
ChangeStreamIterable<Document> iterable = db.getCollection("products").watch()
.fullDocument(FullDocument.UPDATE_LOOKUP);
try (MongoChangeStreamCursor<ChangeStreamDocument<Document>> cursor = iterable.cursor()) {
while (cursor.hasNext()) {
ChangeStreamDocument<Document> change = cursor.next();
String docId = change.getDocumentKey().getObjectId("_id").toString();
switch (change.getOperationType()) {
case INSERT, REPLACE, UPDATE -> {
Document doc = change.getFullDocument();
esClient.index(i -> i.index("products").id(docId).document(doc));
}
case DELETE -> esClient.delete(d -> d.index("products").id(docId));
}
saveTokenToRedis(change.getResumeToken());
}
}
4.4 跨区域复制与数据迁移
对于需要跨可用区或跨云同步的场景，Change Streams 可作为轻量级复制管道：
MongoDB Cluster A (Region A)
│
├── Change Streams ──→ 同步程序 ──→ MongoDB Cluster B (Region B)
同步程序负责：
- 读取源端 Change Stream；
- 将
insert/update/replace映射为目标端写操作； - 将
delete映射为目标端删除； - 通过 Resume Token 保证断点续传。
5. Change Streams 性能优化
5.1 batch size 调优
batchSize 控制一次网络往返传输的事件数量。过小的 batch size 导致频繁网络往返；过大则增加客户端内存压力和延迟。建议根据事件大小和下游处理能力动态调整，通常初始值设为 100-1000。
db.getCollection("events").watch().batchSize(500);
5.2 投影减少数据传输
由于 Change Stream 事件默认包含 fullDocument，对于大文档场景会产生大量网络 IO。应使用 $project 裁剪不必要的字段。
const pipeline = [
{ $project: { "fullDocument.content": 0, "updateDescription": 0 } }
];
db.articles.watch(pipeline);
5.3 高吞吐场景的分片策略
在分片集群上，Change Stream 需要聚合多个分片的 oplog。可以通过在 $match 阶段加入片键过滤，让 MongoDB 路由到特定分片，减少聚合开销。
const pipeline = [
{ $match: { "documentKey.shardKey": { $gte: 0, $lt: 100 } } }
];
db.orders.watch(pipeline);
另外，极端高吞吐场景下，可考虑按集合拆分多个 Change Stream 消费者实例，各自负责一部分片键范围，实现水平扩展。
6. Spring Data MongoDB Reactive 集成（WebFlux + Change Streams）
Spring Data MongoDB Reactive 提供了对 Change Streams 的原生支持，可与 WebFlux 结合构建响应式实时推送服务。
6.1 依赖
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-mongodb-reactive</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-webflux</artifactId>
</dependency>
6.2 服务层监听 Change Stream
@Service
public class OrderChangeStreamService {
@Autowired
private ReactiveMongoTemplate mongoTemplate;
public Flux<ChangeStreamEvent<Document>> streamOrderChanges() {
ChangeStreamOptions options = ChangeStreamOptions.builder()
.filter(Aggregation.match(Criteria.where("operationType").in("insert", "update")))
.fullDocumentLookup(FullDocument.UPDATE_LOOKUP)
.build();
return mongoTemplate.changeStream("orders", options, Document.class)
.doOnNext(event -> {
System.out.println("Change: " + event.getOperationType());
});
}
}
6.3 WebFlux SSE 实时推送
@RestController
public class OrderStreamController {
@Autowired
private OrderChangeStreamService streamService;
@GetMapping(value = "/orders/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public Flux<ServerSentEvent<String>> streamOrders() {
return streamService.streamOrderChanges()
.map(event -> ServerSentEvent.builder(event.getRaw().toJson()).build());
}
}
客户端通过 EventSource 连接 /orders/stream 即可实时接收变更推送。
6.4 持久化 Resume Token
在响应式流中保存 Token 需要利用 doOnNext 或自定义 Subscriber：
private volatile BsonDocument lastResumeToken;
public Flux<ChangeStreamEvent<Document>> streamWithRecovery() {
ChangeStreamOptions options = ChangeStreamOptions.builder()
.resumeToken(lastResumeToken)
.build();
return mongoTemplate.changeStream("orders", options, Document.class)
.doOnNext(event -> lastResumeToken = event.getRaw().getResumeToken());
}
生产环境中建议将 lastResumeToken 定期异步写入 Redis 或数据库，避免进程重启导致数据丢失。
7. 事务边界与 Change Streams
7.1 事务内 events 的 clusterTime 一致性
MongoDB 4.0+ 支持多文档事务（Replica Set），4.2+ 支持分片事务。当事务提交时，其中涉及的所有变更事件会共享相同的 clusterTime 和 txnNumber，并且通过 lsid（Logical Session ID）关联。
{
"operationType": "insert",
"clusterTime": { "$timestamp": { "t": 1692000100, "i": 1 } },
"txnNumber": 5,
"lsid": { "id": { "$binary": "...", "$type": "04" }, "uid": "..." }
}
同一事务内的所有事件 clusterTime 相同，txnNumber 递增。这意味着：
- 下游消费者可以通过
lsid+txnNumber将同一事务的事件分组处理。 - 若事务回滚，对应的 Change Stream 事件 不会 被推送。
7.2 事务事件的消费策略
Map<Lsid, List<ChangeStreamDocument<Document>>> txnBuffer = new HashMap<>();
while (cursor.hasNext()) {
ChangeStreamDocument<Document> change = cursor.next();
if (change.getLsid() != null && change.getTxnNumber() != null) {
txnBuffer.computeIfAbsent(change.getLsid(), k -> new ArrayList<>()).add(change);
// 实际生产中需根据后续事件判断事务是否完整
if (isTransactionComplete(change)) {
processTransactionBatch(txnBuffer.remove(change.getLsid()));
}
} else {
process(change);
}
}
注意：事务事件通常不携带 fullDocument（即使配置了 updateLookup），需要通过 updateDescription 自行推断变更内容。
8. 生产监控与告警
8.1 MongoDB Atlas Change Streams 指标
MongoDB Atlas 监控面板提供以下 Change Stream 相关指标：
| 指标 | 含义 | 告警阈值建议 |
|---|---|---|
oplog GB/Hour | oplog 生成速率 | 突增时检查是否有大批量写入 |
Change Stream Open Cursors | 当前活跃的游标数量 | 超过预期值时检查消费者泄漏 |
[Replication](/posts/database/) Lag | Secondary 落后 Primary 的时间 | > 30s 告警，可能影响 Resume Token 窗口 |
Connections | 总连接数 | 接近阈值时扩容或优化连接池 |
8.2 自定义 Prometheus Exporter
对于自建 MongoDB 集群，可以编写自定义 exporter 暴露 Change Stream 消费者健康状态。
from prometheus_client import Gauge, start_http_server
import time
change_stream_lag = Gauge('mongodb_change_stream_lag_seconds',
'Lag between clusterTime and consumer processing time',
['collection'])
change_stream_cursor_count = Gauge('mongodb_change_stream_open_cursors',
'Number of open change stream cursors')
def monitor():
while True:
# 示例：通过 MongoDB 命令获取当前 oplog 时间戳
# 并与消费者最后处理的 clusterTime 对比计算 lag
lag = calculate_lag()
change_stream_lag.labels(collection='orders').set(lag)
change_stream_cursor_count.set(get_open_cursor_count())
time.sleep(15)
if __name__ == '__main__':
start_http_server(9090)
monitor()
配合 Prometheus + Grafana ，可以可视化 Change Stream 消费延迟、断路器状态、重连次数等关键指标，实现生产级可观测性。
延伸阅读
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。