MongoDB Change Streams 要求版本 ≥ 3.6、运行于副本集或分片集群模式，Node.js 驱动需 ≥ 3.1.0（推荐 ≥ 4.0.0），watch() 默认从当前 oplog 位置开始监听，支持 resumeAfter 断点续传及聚合管道过滤事件。
Change Streams 要求 MongoDB 版本和副本集配置
Node.js 中使用 changeStream 前，MongoDB 必须是 3.6+ 且运行在副本集（Replica Set）或分片集群（Sharded Cluster）模式下——单机 mongod 直接启动不支持。本地开发时别用 mongod --dbpath ./data 启动，得用 mongod --replSet rs0 --dbpath ./data，再执行 rs.initiate() 初始化副本集。
驱动版本也得匹配：MongoDB Node.js Driver ≥ 3.1.0（推荐 ≥ 4.0.0），否则 collection.watch() 会静默失败或抛 MongoServerSelectionError。
- 检查副本集状态：连接后运行
db.isMaster()，确认ismaster: true且hosts字段存在 - Driver 版本验证：
npm list mongodb，避免混用mongodb@3.x和@mongo-js/driver - 云服务（如 MongoDB Atlas）默认启用了副本集，但免费层可能限制变更流并发数，注意配额
watch() 的基础用法和 resumable 机制
collection.watch() 返回一个可监听的游标，不是一次性查询。它默认从“当前最新 oplog 时间点”开始监听，意味着启动前发生的变更不会被收到——这点常被误认为“丢数据”，其实不是 bug，是设计如此。
要实现断线重连后的续读，必须传入 resumeAfter 或 startAfter。最稳妥的做法是持久化上一次成功处理的 _id（即 change._id），并在重建流时作为 resumeAfter 传入：
const resumeToken = await getLastProcessedToken(); // 从 Redis 或本地文件读
const changeStream = collection.watch([], {
resumeAfter: resumeToken,
maxAwaitTimeMS: 5000
});
changeStream.on('change', async change => {
await processChange(change);
await saveLastProcessedToken(change._id); // 写回 token
});
-
resumeAfter是精确恢复点，要求 token 必须存在；startAfter允许跳过已丢失的变更，适合容忍少量丢失的场景 -
maxAwaitTimeMS控制服务器端最长等待时间，设太小会导致频繁空轮询，太大则延迟升高 - 不要在
change事件回调里直接await长耗时操作（如 HTTP 请求），应加队列或限流，否则阻塞流接收
过滤变更类型与字段投影的常见写法
用聚合管道数组作为 watch() 第一个参数，能高效过滤事件。比如只监听 insert 和 update，且仅当 status 字段被修改时触发：
collection.watch([
{ $match: {
$or: [{ operationType: 'insert' }, { operationType: 'update' }],
'updateDescription.updatedFields.status': { $exists: true }
}
}
])
注意：updateDescription 只在 update 类型中存在；fullDocument 默认不返回，需显式开启 { fullDocument: 'updateLookup' } 才能拿到更新后的完整文档（含未修改字段）。
-
operationType可取值：insert、update、replace、delete、drop、rename、dropDatabase、invalidate -
fullDocument: 'required'仅在事务中有效，普通写入请用'updateLookup' - 避免在
$match中使用$text或复杂索引字段，变更流不走索引，纯内存过滤
错误处理与连接生命周期管理
Change Stream 不是长连接保活协议，网络抖动、主节点切换、oplog 截断都会导致 error 或 close 事件。不能只监听 change，必须同时处理 error 和 close，并主动重建流：
function createChangeStream() {
const stream = collection.watch(...);
stream.on('change', handle);
stream.on('error', err => {
console.error('ChangeStream error:', err.name, err.message);
if (err.name === 'MongoNetworkError' || err.name === 'MongoCursorExhaustedError') {
setTimeout(createChangeStream, 1000); // 指数退避更佳
}
});
stream.on('close', () => console.log('Stream closed, reconnecting...'));
return stream;
}
-
MongoCursorExhaustedError表示 oplog 已被清理（如oplogSize太小或停机太久），此时resumeAftertoken 失效，需降级为全量同步或报错退出 - 应用进程退出前，务必调用
stream.close()，否则可能泄漏 socket 和内存 - 多个集合监听时，别共用一个 client 连接池却各自开流——每个
watch()都会占用一个 socket，连接池maxPoolSize得相应调大
变更流本身不保证 exactly-once，resumeAfter 也无法跨分片精确对齐；高可靠场景下，仍建议以变更事件为触发器，再通过 _id 查库做幂等校验。
