MongoDB的Change Stream为实时数据同步提供了极大便利，但生产环境中经常出现这样的场景：监听程序运行得好好的，突然日志里抛出一串错误信息，其中code: 1290格外扎眼，变更流随之断开，后续的数据变更再也没有被消费到。错误码1290在MongoDB中对应OperationFailed类别，当它出现在变更流场景时，核心含义通常是服务端认为当前这条变更流已经无法继续恢复，客户端携带的resumeToken已经失效。这篇文章就来详细拆解1290错误背后的成因、resumeToken的工作机制，以及断开后如何正确重连。
错误码1290的底层原因：为什么resumeToken会失效
要理解1290错误，先要明白变更流的实现原理。Change Stream本质上建立在oplog（操作日志）之上，客户端每次打开变更流时，服务端会返回一个resumeToken，本质上是oplog中某条记录的位置标识。当连接意外断开后，客户端携带resumeToken重新发起aggregate命令，服务端就从该位置之后继续推送变更。
这套机制的前提是resumeToken指向的oplog记录仍然存在。MongoDB副本集的oplog是一个固定大小的集合（capped collection），空间用完后旧记录会被滚动覆盖。如果监听程序断开的时间过长，或者业务写入量极大导致oplog窗口很短，resumeToken指向的位置可能早已被覆盖，此时服务端就无法从该位置恢复，直接返回1290错误，并在响应中标记resumable: false。
如何确认和排查1290错误的具体诱因
拿到1290错误后不要急着盲目重试，盲目重试只会不断收到同样的错误。排查的第一步是打印完整的错误对象，重点看code和errmsg字段：
// Node.js 驱动中捕获变更流错误
collection.watch([]).on('error', (err) => {
console.error('错误码:', err.code);
console.error('错误信息:', err.errmsg || err.message);
console.error('是否可恢复:', err.code === 1290);
});其次检查oplog的窗口时长。登录副本集成员，查询oplog中最早记录的时间戳与当前时间的差值，如果这个差值很小（比如只有几分钟），说明oplog滚动很快，任何一次超过该时长的断开都会导致1290。可以适当增大oplog大小，或者排查是否存在异常大量的写入。
另外两类常见诱因也需要排除：一是被监听的集合被删除后重建，resumeToken对应的namespace已不存在；二是副本集发生了主从切换或成员拓扑变化，虽然驱动通常会自动处理这类场景，但如果切换期间服务端版本差异较大或token跨节点无法定位，同样会触发1290。
断开重连的三种处理方案对比
方案一：自动重连。对于网络抖动、瞬时故障，Node.js驱动内置的tryNext与自动重试机制基本可以覆盖，程序只需要监听error事件后重新调用watch并传入上一次保存的resumeToken即可。这种方式实现最简单，适合绝大多数短暂断开的场景。
方案二：手动恢复。持久化每次变更事件中的_id（即resumeToken）到数据库或文件，程序重启或断开后从存储中读取token，通过resumeAfter参数恢复。这是生产环境最推荐的方式，代码示例如下：
const { MongoClient } = require('mongodb');
async function startWatch(client, collection) {
// 从外部存储读取上次保存的 token，首次运行为 null
const lastToken = await loadTokenFromStore();
const changeStream = collection.watch(
[{ $match: { operationType: { $in: ['insert', 'update', 'replace'] } } }],
lastToken ? { resumeAfter: lastToken } : {}
);
changeStream.on('change', async (event) => {
// 每消费一条变更就持久化 token，保证断点可恢复
await saveTokenToStore(event._id);
console.log('捕获变更:', event.documentKey);
});
changeStream.on('error', async (err) => {
console.error('变更流错误:', err.code, err.message);
if (err.code === 1290) {
// token 已失效，只能从当前时刻重新开始监听
await saveTokenToStore(null);
}
// 关闭旧流，延迟后重建
changeStream.close();
setTimeout(() => startWatch(client, collection), 3000);
});
}方案三：全量重新初始化。当token失效且业务不能容忍数据丢失时，需要对集合做一次全量扫描补偿断开期间的变更，再从当前时间点开始新的监听。可以配合startAtOperationTime指定从某个时间点开始，但要注意该时间点同样受oplog窗口限制，超出了窗口一样会失败。这种方案代价最大，通常作为数据一致性兜底手段。
生产环境中的预防措施与常见坑点
预防永远优于补救。第一，合理规划oplog大小，通过db.adminCommand({replSetResizeOplog: 1, size: 20480})可以动态调整oplog容量，确保窗口时长覆盖可能的最大断开时间，比如预留几个小时甚至一天。第二，消费逻辑要尽量轻量，把耗时的业务处理放到独立队列中异步执行，避免变更流因处理阻塞而长时间未拉取数据。
坑点方面，resumeAfter和startAtOperationTime不能同时指定，同时传入会直接报参数冲突。另外token必须取自变更事件中的_id字段，而不是文档本身的_id，两者结构完全不同。还有一点容易被忽略：每次change事件都要更新持久化的token，如果只在程序退出前保存一次，进程被强杀时就会丢失最后的进度，下一次恢复时可能重复消费，需要业务侧做好幂等处理。
总结来说，1290错误并不可怕，它只是服务端明确告知你这条变更流的恢复点已经不存在了。理解oplog与resumeToken的关系，做好token的持久化与幂等消费，再配合指数退避的重连策略，就能构建出一条健壮的变更流监听链路。
MongoDB故障码1290变更流断开重连Change Stream修改时间：2026-09-01 05:22:50