非分片非副本集MongoDB如何不使用.watch()监听集合增改操作

你之前用aggregate的写法行不通，因为aggregate是用来对已有数据做聚合分析的工具，只会一次性返回符合条件的历史数据，不会持续监听新产生的变更，自然不会触发data事件捕获后续的插入/更新操作。

针对无法修改MongoDB配置（不能搭建副本集）的情况，给你三个实用的替代方案：

方案1：用Mongoose中间件在应用层拦截操作 #

这是最直接的方式，利用Mongoose的中间件钩子，在数据插入、更新的前后触发自定义逻辑，实时捕获当前应用发起的所有写入操作。

需要覆盖不同类型的写入操作（文档实例操作和模型级操作），代码示例：

// 针对save、create这类文档实例操作
Schema.pre('save', function(next) {
  const operationType = this.isNew ? 'insert' : 'update';
  console.log(`捕获到${operationType}操作，文档内容:`, this.toObject());
  next();
});

// 针对updateOne、updateMany这类模型级批量操作
Schema.pre('updateOne', function(next) {
  console.log('捕获到update操作，查询条件:', this.getQuery(), '更新内容:', this.getUpdate());
  next();
});

// 针对findOneAndUpdate这类原子更新操作
Schema.pre('findOneAndUpdate', function(next) {
  console.log('捕获到findOneAndUpdate操作，查询条件:', this.getQuery(), '更新内容:', this.getUpdate());
  next();
});

优点：实时性高，无额外查询开销；缺点：只能捕获当前Mongoose实例发起的操作，其他客户端直接写入数据库的行为监听不到。

方案2：定时轮询对比数据 #

如果需要监听所有来源的写入操作（包括其他客户端），可以用定时轮询的方式，定期查询集合中新增或更新的文档。

实现思路 #

• 给集合添加updatedAt字段（建议默认自动更新），或者利用_id的时间戳（MongoDB的ObjectId前4字节是时间戳）
• 记录每次轮询的最后时间戳，每次只查询时间戳大于该值的文档

代码示例（用updatedAt字段）：

let lastCheckTime = new Date();

async function pollCollection() {
  try {
    const changes = await Model.find({
      updatedAt: { $gt: lastCheckTime }
    }).sort({ updatedAt: 1 });

    if (changes.length > 0) {
      changes.forEach(doc => {
        const operationType = doc.createdAt.getTime() === doc.updatedAt.getTime() ? 'insert' : 'update';
        console.log(`捕获到${operationType}操作，文档:`, doc.toObject());
      });
      // 更新最后检查时间为最新的文档更新时间
      lastCheckTime = changes[changes.length - 1].updatedAt;
    }
  } catch (error) {
    console.error('轮询出错:', error);
  }
}

// 每5秒轮询一次，可根据需求调整间隔
setInterval(pollCollection, 5000);

优点：能监听所有客户端的写入操作；缺点：有延迟（取决于轮询间隔），会产生额外的查询开销。

方案3：自定义操作日志集合 #

如果你的应用是唯一的写入来源，可以在每次写入业务数据时，同时把操作记录到一个专门的日志集合，再轮询这个日志集合获取变更，这种方式比直接轮询业务集合更轻量。

代码示例：

// 1. 定义操作日志的Schema和Model
const OperationLogSchema = new mongoose.Schema({
  collection: String,
  operationType: { type: String, enum: ['insert', 'update'] },
  documentId: mongoose.Schema.Types.ObjectId,
  timestamp: { type: Date, default: Date.now },
  data: mongoose.Schema.Types.Mixed
});
const OperationLog = mongoose.model('OperationLog', OperationLogSchema);

// 2. 在业务集合的中间件中记录日志
Schema.post('save', async function(doc) {
  const operationType = this.isNew ? 'insert' : 'update';
  await OperationLog.create({
    collection: 'yourCollectionName',
    operationType,
    documentId: doc._id,
    data: doc.toObject()
  });
});

// 3. 轮询日志集合获取变更
let lastLogTime = new Date();
async function pollLogs() {
  const logs = await OperationLog.find({
    timestamp: { $gt: lastLogTime }
  }).sort({ timestamp: 1 });
  
  logs.forEach(log => {
    console.log(`捕获到${log.operationType}操作，文档ID:`, log.documentId);
  });
  
  if (logs.length > 0) {
    lastLogTime = logs[logs.length - 1].timestamp;
  }
}
setInterval(pollLogs, 3000);

优点：逻辑清晰，日志集合数据量小，查询效率高；缺点：同样只能捕获当前应用的写入操作。

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