Change Streams 必须运行在副本集或分片集群上,单节点不支持;需显式传 resumeAfter 保证不丢事件;监听器断开需手动重建并指数退避重试;驱动需 v3.6+ 且聚合管道仅支持特定阶段。

Change Streams 必须运行在副本集或分片集群上
单节点 MongoDB 实例默认不支持 Change Streams,直接调用 collection.watch() 会报错:MongoError: This operation is not supported in a standalone configuration。不是代码写错了,是部署模式不符合要求。
解决方法只有两个:
- 本地开发时,用
mongod --replSet rs0启动副本集,并执行rs.initiate() - 生产环境务必使用副本集(至少 1 主 2 副)或分片集群,不能用单节点
跳过这步,后续所有监听逻辑都会静默失败或抛异常,且错误提示不直观。
collection.watch() 的 filter 和 resumeAfter 参数怎么设才安全
默认调用 collection.watch() 会从当前最新 oplog 位置开始监听,漏掉启动前的变更。要保证不丢事件,必须显式传入 resumeAfter 或 startAfter。
常见做法是首次监听时用空 filter + 记录 resumeToken,后续重启时带上它:
const changeStream = collection.watch([], {
resumeAfter: lastResumeToken // 上次断开时保存的 token
});
changeStream.on('change', (change) => {
console.log(change.fullDocument); // 插入/更新后的完整文档
lastResumeToken = change._id; // 持久化这个 token(例如写入 Redis 或文件)
});
注意:filter 是 BSON 查询对象,比如只监听插入: [{ $match: { operationType: 'insert' } }];不要传 JavaScript 对象字面量,否则无效。
监听器意外断开后如何自动重连
Change Streams 不会自动重连。网络抖动、主节点切换、oplog 被覆盖都可能导致 changeStream.on('close') 或 'error' 触发,此时流就永久终止了。
必须手动重建流,且要避免无限重试压垮服务:
- 监听
'error'和'close'事件,在回调里调用新collection.watch() - 加退避重试(如指数退避),首次延迟 100ms,失败后翻倍,上限 5s
- 检查
change._id是否为null—— 若是,说明 oplog 已被轮转覆盖,需回退到更早时间点或接受丢数据
别依赖 maxAwaitTimeMS 来“保活”,它只控制单次请求等待时长,不解决连接生命周期问题。
Node.js 驱动版本和聚合管道的兼容性坑
Change Streams 在 Node.js 驱动 v3.6+ 才稳定支持全功能,v4.x 起强制要求 MongoDB 4.2+ 服务端。如果用旧版驱动(比如 v3.2),watch() 可能返回空流或静默失败。
另外,聚合管道阶段有严格限制:
- 只允许
$match、$project、$addFields、$replaceRoot、$redact - 不能用
$lookup、$sort、$limit—— 这些会直接导致commandFailed错误 - filter 中字段名必须带引号,例如
{ "operationType": "update" },写成{ operationType: "update" }在某些驱动版本下会被忽略
最稳妥的做法:先用空 filter 测试通路,再逐步加 $match 条件验证行为是否符合预期。


















