如何在MongoDB中监听变更流并获取AJAX两次请求间的更新？

嘿，我刚好做过类似的需求，给你一步步拆解怎么实现用MongoDB 3.6的watch()特性，让前端AJAX定期获取两次调用之间的数据库更新，并且正确恢复变更流！

核心思路 #

核心就是利用MongoDB变更流的resumeToken来标记上次获取变更的位置：

• 前端第一次请求时不带resumeToken，后端从头开始拉取变更
• 后端每次返回变更时，同时返回最新的resumeToken
• 前端把这个token存在本地（比如localStorage），下次请求时带上
• 后端用这个token初始化变更流，就能从上次中断的位置继续获取新增的变更

后端实现（Node.js + Express为例） #

先看后端的代码，这里用MongoDB官方驱动来处理变更流：

const express = require('express');
const MongoClient = require('mongodb').MongoClient;
const app = express();
const port = 3000;

let db;

// 连接MongoDB数据库
MongoClient.connect('mongodb://localhost:27017', { 
  useNewUrlParser: true, 
  useUnifiedTopology: true 
})
.then(client => {
  db = client.db('你的数据库名');
  console.log('成功连接MongoDB');
})
.catch(err => console.error('MongoDB连接失败:', err));

// 处理变更获取的API接口
app.get('/api/get-changes', async (req, res) => {
  try {
    const targetCollection = db.collection('你的集合名');
    // 解析前端传来的resumeToken（如果有的话）
    const resumeToken = req.query.resumeToken ? JSON.parse(req.query.resumeToken) : null;

    // 配置变更流选项
    const streamOptions = {
      fullDocument: 'updateLookup' // 这个选项能让我们拿到更新后的完整文档，非常实用
    };
    // 如果有resumeToken，就设置从该位置恢复
    if (resumeToken) {
      streamOptions.resumeAfter = resumeToken;
    }

    // 创建变更流
    const changeStream = targetCollection.watch([], streamOptions);

    const collectedChanges = [];
    let latestResumeToken = resumeToken;

    // 监听变更事件，收集所有新变更
    changeStream.on('change', (change) => {
      collectedChanges.push(change);
      latestResumeToken = change._id; // 每次变更的_id就是最新的resumeToken
    });

    // 等待1秒收集当前可用的变更（适配定期AJAX的短连接场景）
    await new Promise(resolve => setTimeout(resolve, 1000));

    // 关闭变更流，释放资源
    changeStream.close();

    // 返回结果给前端
    res.json({
      changes: collectedChanges,
      resumeToken: latestResumeToken
    });
  } catch (err) {
    console.error('获取变更失败:', err);
    // 如果resumeToken无效（比如过期），返回空token让前端从头开始
    res.status(500).json({
      changes: [],
      resumeToken: null,
      error: '变更流恢复失败，将从头开始获取'
    });
  }
});

app.listen(port, () => {
  console.log(`服务器运行在端口 ${port}`);
});

前端AJAX实现 #

前端需要定期发送请求，保存resumeToken，并处理返回的变更：

// 从本地存储读取上次的resumeToken
let lastResumeToken = localStorage.getItem('dbResumeToken') 
  ? JSON.parse(localStorage.getItem('dbResumeToken')) 
  : null;

// 每5秒调用一次API获取变更（时间间隔可以根据你的需求调整）
setInterval(async () => {
  try {
    const queryParams = new URLSearchParams();
    if (lastResumeToken) {
      queryParams.append('resumeToken', JSON.stringify(lastResumeToken));
    }

    const response = await fetch(`/api/get-changes?${queryParams}`);
    const result = await response.json();

    // 处理错误情况
    if (result.error) {
      console.error(result.error);
      lastResumeToken = null;
      localStorage.removeItem('dbResumeToken');
      return;
    }

    // 处理获取到的变更数据
    if (result.changes.length > 0) {
      console.log('收到新的数据库变更:', result.changes);
      // 这里根据不同的操作类型处理UI更新
      result.changes.forEach(change => {
        switch (change.operationType) {
          case 'insert':
            console.log('新增文档:', change.fullDocument);
            // 写你的新增UI逻辑
            break;
          case 'update':
            console.log('更新后的文档:', change.fullDocument);
            // 写你的更新UI逻辑
            break;
          case 'delete':
            console.log('删除的文档ID:', change.documentKey._id);
            // 写你的删除UI逻辑
            break;
          // 其他操作类型比如replace可以按需处理
        }
      });
    }

    // 更新并保存最新的resumeToken
    if (result.resumeToken) {
      lastResumeToken = result.resumeToken;
      localStorage.setItem('dbResumeToken', JSON.stringify(lastResumeToken));
    }
  } catch (err) {
    console.error('AJAX请求失败:', err);
  }
}, 5000);

关键注意点 #

• resumeToken的有效性：MongoDB会保留变更流的历史一段时间（默认是24小时），如果超过这个时间没请求，resumeToken会失效，这时候后端会报错，前端需要重置token从头开始。
• fullDocument选项：默认情况下，update类型的变更只会返回更新的字段，设置fullDocument: 'updateLookup'可以拿到完整的更新后文档，对前端UI更新更友好。
• 短连接 vs 长连接：这里用的是定期AJAX的短连接方式，如果你的需求是实时推送，其实可以用WebSocket代替AJAX，让后端保持变更流的长连接，有变更就主动推给前端，但你的需求是定期调用，所以短连接更合适。

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