如何实现MongoDB插入满100条文档时触发事件供JS批量处理

针对你要的「持续插入文档、累计到指定数量自动触发、JS读取批次处理」的需求，按入门难度从低到高给3种可直接落地的路径，新手可以从最简单的开始试：

方案1：定时轮询（零高级特性依赖，最适合刚上手的新手） #

这个方案对MongoDB版本、部署模式没有任何要求，单实例、副本集都能跑，逻辑直白好排查：

• 业务文档正常往集合插入，每条文档插入时默认带上processed: false标记，同时保留MongoDB默认生成的_id（默认ObjectId本身带插入时间戳，天然支持按插入顺序排序）
• 写一个常驻运行的JS脚本，按固定间隔（比如每5秒、每10秒，根据你对实时性的要求调整）执行一次检查：统计集合中processed: false的文档总数，当总数大于等于你设的阈值（比如100），就按_id升序取最早的100条未处理文档
• 拿到批次文档后执行自定义处理逻辑，处理完成后批量给这100条文档更新processed: true标记，等待下一轮检查
• 兜底逻辑：脚本重启时先查一次有没有凑够数量的未处理文档，避免服务重启漏处理批次

方案2：Change Streams 实时监听（实时性最好，JS生态适配度最高） #

如果你需要凑够阈值立刻触发，不想等轮询间隔，可以用MongoDB原生的变更流能力，MongoDB 3.6及以上版本支持，开发环境单实例只要开启副本集模式就能用，不需要额外装组件：

• 单独建一个计数辅助集合，比如叫batch_tracker，专门存当前批次的累计数量、批次起始文档ID、处理状态，初始化一条记录存目标集合名、初始计数0、收集状态
• 用MongoDB JS驱动打开目标集合的变更流，只监听insert类型的事件，每收到一条新插入的文档通知，就对辅助集合里的计数做原子自增，第一次自增时顺便把当前文档的_id存为批次起始ID
• 每次自增后检查计数，当数值刚好等于阈值时，先把批次状态改成「处理中」避免重复触发，再拉取从起始ID开始的100条未处理文档，执行批量逻辑
• 处理完成后给这100条文档打已处理标记，同时把辅助集合的计数重置为0、状态切回「收集中」，开始累计下一批

最简Node.js实现参考：

const { MongoClient } = require('mongodb');
const THRESHOLD = 100;
const COL_NAME = '你的业务集合名';
const DB_CONN = '你的MongoDB连接串';
const DB_NAME = '你的库名';

async function runBatchProcess(docs) {
  // 这里替换成你自己的批量处理逻辑
  console.log(`处理批次，共${docs.length}条文档`);
}

async function start() {
  const client = await MongoClient.connect(DB_CONN);
  const db = client.db(DB_NAME);
  const bizCol = db.collection(COL_NAME);
  const trackerCol = db.collection('batch_tracker');

  // 初始化计数记录
  await trackerCol.updateOne(
    { targetCol: COL_NAME },
    { $setOnInsert: { count: 0, startId: null, status: 'collecting' } },
    { upsert: true }
  );

  // 启动先补处理遗留的未完成批次
  const pending = await trackerCol.findOne({ targetCol: COL_NAME, status: 'processing' });
  if (pending && pending.count >= THRESHOLD) {
    const docs = await bizCol.find({ _id: { $gte: pending.startId }, processed: { $ne: true } }).limit(THRESHOLD).toArray();
    await runBatchProcess(docs);
    await bizCol.updateMany({ _id: { $in: docs.map(d => d._id) } }, { $set: { processed: true } });
    await trackerCol.updateOne({ _id: pending._id }, { $set: { count: 0, startId: null, status: 'collecting' } });
  }

  // 开启变更流监听
  const stream = bizCol.watch([{ $match: { operationType: 'insert' } }]);
  stream.on('change', async (change) => {
    const docId = change.fullDocument._id;
    const tracker = await trackerCol.findOneAndUpdate(
      { targetCol: COL_NAME, status: 'collecting' },
      { $inc: { count: 1 }, $setOnInsert: { startId: docId } },
      { returnDocument: 'after' }
    );

    if (tracker.count === THRESHOLD) {
      // 锁批次防重复触发
      await trackerCol.updateOne(
        { _id: tracker._id, count: THRESHOLD },
        { $set: { status: 'processing' } }
      );
      const batchDocs = await bizCol.find({ _id: { $gte: tracker.startId }, processed: { $ne: true } }).limit(THRESHOLD).toArray();
      await runBatchProcess(batchDocs);
      await bizCol.updateMany({ _id: { $in: batchDocs.map(d => d._id) } }, { $set: { processed: true } });
      await trackerCol.updateOne(
        { _id: tracker._id },
        { $set: { count: 0, startId: null, status: 'collecting' } }
      );
    }
  });
}
start();

方案3：云服务/企业版自带触发器 #

如果你用的是MongoDB Atlas云服务或者企业版，可以直接用平台自带的触发器功能，在控制台配置插入监听规则、触发阈值，关联写好的JS处理函数就行，不需要自己维护监听服务。缺点是社区版自建MongoDB不支持，学习阶段不推荐，容易被平台绑定。

新手避坑提示：所有计数更新一定要用MongoDB的原子更新操作，不要先查计数、改完再写回去，并发插入的时候会出现计数不准、批次漏数重复的问题。不要把处理逻辑写在数据库端的存储过程里，调试排错成本极高，所有业务逻辑放在JS应用层实现，后续改逻辑、扩性能都方便。

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