如何确保MongoDB变更流事件在GCP Cloud Run多实例下仅被消费一次？

针对你在GCP Cloud Run多NodeJS实例下，重复接收MongoDB TTL文档删除事件的问题，以下是几个靠谱的解决思路：

1. 用MongoDB消费者组（推荐，Atlas原生支持） #

MongoDB Atlas的变更流支持消费者组功能，能自动把变更事件分发给组内的单个消费者，避免重复投递。每个实例在组内有唯一标识，Atlas会跟踪每个消费者的进度，确保同一件事只被一个实例处理。

代码示例：

const { MongoClient } = require('mongodb');

async function watchTTLDeletes() {
  const client = await MongoClient.connect(process.env.MONGODB_URI);
  const db = client.db('your-db-name');
  const collection = db.collection('your-collection');

  // 生成当前实例的唯一消费者ID，用Cloud Run环境变量加随机串确保唯一性
  const consumerId = `cloud-run-${process.env.K_SERVICE}-${process.env.K_REVISION}-${Math.random().toString(36).slice(2, 10)}`;

  const changeStream = collection.watch(
    [{ $match: { operationType: 'delete' } }], // 只监听删除事件
    {
      consumerGroupId: 'ttl-delete-event-group', // 统一的消费者组ID
      consumerId: consumerId,
      maxAwaitTimeMS: 15000
    }
  );

  changeStream.on('change', async (event) => {
    try {
      // 这里写你的删除事件处理逻辑
      console.log(`处理事件: ${event.documentKey._id}，实例: ${consumerId}`);
      // 处理完成后，MongoDB会自动更新该消费者的resume位置，无需手动处理
    } catch (err) {
      console.error('处理事件失败:', err);
      // 若处理失败，可根据情况决定是否关闭流或重试
    }
  });
}

watchTTLDeletes();

2. 分布式锁+幂等校验 #

如果不想用消费者组，可以给每个事件加锁，确保只有抢到锁的实例能处理。用MongoDB本身的原子操作实现乐观锁：

1. 创建专门的event_processing集合，用来记录事件的处理状态
2. 每个实例收到事件后，先尝试将事件标记为“已处理中”，只有成功标记的实例才继续处理

代码示例：

async function handleDeleteEvent(event, db) {
  const eventId = event.documentKey._id;
  // 用findOneAndUpdate做原子操作，尝试抢锁
  const lockResult = await db.collection('event_processing').findOneAndUpdate(
    { eventId: eventId, processed: false },
    { 
      $set: { 
        processed: true, 
        processedBy: process.env.K_REVISION, 
        processedAt: new Date() 
      } 
    },
    { upsert: true, returnDocument: 'after' }
  );

  // 只有当前实例成功将状态改为processed的，才处理事件
  if (lockResult.value.processedBy === process.env.K_REVISION) {
    // 执行你的删除事件处理逻辑
    console.log(`实例 ${process.env.K_REVISION} 处理事件 ${eventId}`);
  } else {
    // 跳过，事件已经被其他实例处理了
    console.log(`事件 ${eventId} 已被其他实例处理，跳过`);
  }
}

3. 引入GCP Pub/Sub做中间层 #

把变更流的事件先转发到Pub/Sub，再让Cloud Run实例订阅Pub/Sub主题。Pub/Sub本身支持消息确认机制，只有实例调用ack()后，消息才会被移除；如果处理失败，Pub/Sub会重试。同时结合业务层的幂等校验，确保重复消息不会重复执行逻辑：

1. 写一个单独的服务（比如Cloud Function）监听MongoDB变更流，把删除事件发送到Pub/Sub主题
2. Cloud Run实例订阅该主题，收到消息后先检查事件ID是否已经处理过（比如存在数据库或缓存里），没处理过再执行逻辑，处理完成后调用ack()

额外提醒：业务逻辑要做幂等 #

不管用哪种方案，都建议让你的事件处理逻辑本身支持幂等。比如处理删除事件时，即使重复收到，执行多次也不会造成数据异常（比如检查关联资源是否已经被清理，已清理就直接返回）。

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