Mongo watch() change stream无报错突然停止工作如何排查解决

以下是同技术栈生产环境踩中同类问题的核心诱因，对应MongoDB Atlas三节点集群+Mongoose的使用场景：

• 主节点选举/轮换静默断连：Atlas集群会定期自动打补丁、做主从节点切换，这个过程中旧的change stream长连接会被断开，但Mongoose客户端侧不会触发error事件，如果没监听close/end事件，代码层面完全感知不到，直接进入空闲状态。
• 闲置游标被回收：如果监听的集合连续30分钟以上没有任何变更，Atlas会主动回收闲置的change stream游标，客户端同样不会抛显式错误，直接停止推送事件。
• 连接参数配置不匹配：默认Mongoose连接配置没有针对长连接监听做优化，心跳间隔过长、连接池预留不足，遇到网络波动的时候连接已经死了，但客户端维护的连接池还认为连接是存活的，不会主动重连。
• 事件监听不全：绝大多数裸写watch()的实现只监听change和error两个事件，而静默断连、游标回收的场景只会触发close或end事件，完全不会进error回调，自然没法触发重连。

第一步 调整Mongoose连接配置 #

连接串和连接参数追加以下配置，适配Atlas集群的长连接场景：

// 连接串末尾追加参数：retryWrites=true&maxIdleTimeMS=300000&connectTimeoutMS=10000
mongoose.connect(MONGO_URI, {
  heartbeatFrequencyMS: 10000, // 10秒发一次心跳，及时感知死连接
  minPoolSize: 2, // 预留长连接名额，避免连接被业务请求占满
  serverSelectionTimeoutMS: 5000,
  socketTimeoutMS: 45000
})

第二步 封装带断点恢复、全事件监听、健康检查的Change Stream逻辑 #

不要直接裸调watch()，参考以下实现做封装，核心是三个点：持久化resume token、监听所有生命周期事件、定时健康检查主动重建异常连接：

let resumeToken = null
let changeStream = null

// 从持久化存储读取上次消费断点，存在Redis/本地文件都可以，不要仅存在内存
async function getStoredResumeToken() {
  return await redisClient.get('mongo:change_stream:resume_token')
}
async function saveResumeToken(token) {
  resumeToken = token
  await redisClient.set('mongo:change_stream:resume_token', token)
  await redisClient.set('mongo:change_stream:last_event_time', Date.now().toString())
}

async function initChangeStream() {
  // 先销毁已存在的异常实例
  if (changeStream) {
    try { await changeStream.close() } catch (err) {}
  }

  const watchOptions = {
    fullDocument: 'updateLookup',
    maxAwaitTimeMS: 60000, // 1分钟长轮询超时，避免游标长期闲置被回收
    batchSize: 100
  }
  // 有断点就从断点续传，没有就从当前时间开始监听
  if (resumeToken) {
    watchOptions.resumeAfter = resumeToken
  }

  // 替换成你要监听的集合，也可以直接监听整个库
  changeStream = TargetCollection.watch([], watchOptions)

  changeStream.on('change', async (event) => {
    await saveResumeToken(event._id)
    // 这里写你的业务事件处理逻辑
    handleChangeEvent(event)
  })

  // 所有异常退出场景都触发延迟重连
  const reconnect = (reason) => {
    console.warn(`change stream异常退出: ${reason}, 5秒后重建`)
    setTimeout(initChangeStream, 5000)
  }
  changeStream.on('error', (err) => {
    // 如果是resume token过期的错误，清空旧token重新从头监听
    if (err.code === 260 || err.message.includes('resume token')) {
      resumeToken = null
      saveResumeToken(null)
    }
    reconnect(`error: ${err.message}`)
  })
  changeStream.on('close', () => reconnect('connect closed'))
  changeStream.on('end', () => reconnect('cursor end'))
}

// 定时健康检查，兜底静默断连的场景
setInterval(async () => {
  try {
    // 检查mongo连接存活
    await mongoose.connection.db.admin().ping()
    // 检查距离上次收到事件是否超过10分钟，超过就主动重建
    const lastEventTime = Number(await redisClient.get('mongo:change_stream:last_event_time') || 0)
    if (Date.now() - lastEventTime > 10 * 60 * 1000) {
      reconnect('long time no event')
    }
  } catch (err) {
    reconnect(`health check failed: ${err.message}`)
  }
}, 60 * 1000)

// 服务启动时初始化
getStoredResumeToken().then(token => {
  resumeToken = token
  initChangeStream()
})

• 先去Atlas集群的事件日志页，对齐change stream停止的时间点，看对应时间有没有主节点选举、自动补丁升级、IP白名单拦截、连接数超限的记录，先排除集群侧的触发原因。
• 本地调试时打开Mongoose debug日志：mongoose.set('debug', { color: true })，看断流时底层连接有没有被静默回收的日志。
• 测试环境可以手动在Atlas控制台触发测试故障转移，验证重连逻辑是否生效，不需要等生产环境偶发。

注意：Atlas默认resume token的有效期是24小时，如果你的服务停机时间超过24小时，旧的resume token会失效，重连时需要捕获对应错误清空token，否则会陷入无限重连失败的循环。

内容的提问来源于stack exchange，提问作者debr