NodeJS与RethinkDB监听表实时变更时如何处理连接中断及重试问题

可行解决思路 #

1. 优先捕获Feed对象抛出的连接异常 #

RethinkDB返回的changefeed Feed实例属于EventEmitter，连接断开时抛出的ReqlRuntimeError会通过error事件传递，只要主动绑定该事件监听即可避免异常向上冒泡导致进程崩溃。
参考实现代码：

const r = require('rethinkdb')
let reconnectTimer = null
const RETRY_DELAY = 5000 // 自定义重连间隔，单位毫秒

async function startChangefeedListen() {
  let conn = null
  let feed = null
  try {
    // 建立数据库连接
    conn = await r.connect({ 
      host: '你的RethinkDB服务地址', 
      port: 28015, 
      db: '目标数据库名' 
    })
    // 启动表变更监听
    feed = await r.table('users').changes().run(conn)
    
    // 绑定feed错误事件，捕获断连异常
    feed.on('error', (err) => {
      console.error('Changefeed运行异常:', err.message)
      // 清理残留资源
      feed.close().catch(() => {})
      conn.close().catch(() => {})
      // 触发延时重连
      clearTimeout(reconnectTimer)
      reconnectTimer = setTimeout(startChangefeedListen, RETRY_DELAY)
    })

    // 正常处理变更数据
    feed.each((err, change) => {
      if (err) throw err
      // 此处编写自定义变更处理逻辑
      console.log('收到表变更数据:', change)
    })

    // 绑定连接关闭事件
    conn.on('close', () => {
      console.log('数据库连接已断开')
      clearTimeout(reconnectTimer)
      reconnectTimer = setTimeout(startChangefeedListen, RETRY_DELAY)
    })
  } catch (err) {
    console.error('启动监听失败:', err.message)
    // 连接建立失败也触发重试逻辑
    clearTimeout(reconnectTimer)
    reconnectTimer = setTimeout(startChangefeedListen, RETRY_DELAY)
  }
}

// 启动监听任务
startChangefeedListen()

2. 全局异常兜底 #

为了避免遗漏监听的异常场景导致进程崩溃，可以添加全局异常捕获逻辑，专门过滤RethinkDB连接类异常，触发重连流程：

process.on('unhandledRejection', (err) => {
  if (err.name === 'ReqlRuntimeError' && err.message.includes('Connection is closed')) {
    console.log('捕获到RethinkDB断连异常，触发重连逻辑')
    clearTimeout(reconnectTimer)
    reconnectTimer = setTimeout(startChangefeedListen, RETRY_DELAY)
    return
  }
  // 其他异常按原有逻辑处理
  console.error('未处理的Promise拒绝:', err)
})

process.on('uncaughtException', (err) => {
  if (err.name === 'ReqlRuntimeError' && err.message.includes('Connection is closed')) {
    console.log('捕获到RethinkDB断连异常，触发重连逻辑')
    clearTimeout(reconnectTimer)
    reconnectTimer = setTimeout(startChangefeedListen, RETRY_DELAY)
    return
  }
  // 非连接类异常建议退出进程，避免业务状态异常
  console.error('未捕获的全局异常:', err)
  process.exit(1)
})

3. 进程守护兜底方案 #

如果接受应用重启的方案，可以使用进程守护工具配置异常退出自动重启：

• 配置重启延迟，避免数据库未恢复时频繁重试
• 配置重试次数阈值，超过阈值后暂停重启并推送告警通知
• 可以搭配健康检查接口，监控应用监听状态是否正常

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