面向多MongoDB集合更新追踪的可靠、容错且可扩展解决方案需求

嘿，针对你需要追踪MongoDB中CouponCartRule和UniqueCoupon集合更新的需求，我整理了一套可靠、容错且可扩展的方案，完全适配这类业务场景：

首先推荐用MongoDB原生的Change Streams来做更新追踪，原因很直观：

• 可靠性：基于MongoDB的oplog实现，不会遗漏任何更新事件；
• 容错性：支持断点续传，即使服务重启也能从上次中断的位置继续监听；
• 扩展性：天然支持副本集和分片集群，后续业务扩容无需重构追踪逻辑。

1. 前置准备：确保MongoDB部署类型 #

Change Streams仅支持副本集或分片集群，单节点部署无法使用，所以先把你的MongoDB升级到副本集或分片架构（这也是生产环境的标配，能同步提升数据可靠性）。

2. 编写Change Stream监听逻辑 #

下面用Node.js的MongoDB驱动示例，展示如何监听两个目标集合的所有关键操作（新增、更新、替换、删除）：

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

// 全局客户端实例，用于事件处理中的数据库操作
let client;

async function setupChangeStreams() {
  const uri = 'mongodb://your-replica-set-node1:27017,your-replica-set-node2:27017,your-replica-set-node3:27017/?replicaSet=your-replica-set-name';
  client = new MongoClient(uri, { retryWrites: true });

  try {
    await client.connect();
    const db = client.db('your-business-db');

    // 监听CouponCartRule集合的核心操作
    const couponRuleStream = db.collection('CouponCartRule').watch(
      [
        { $match: { operationType: { $in: ['insert', 'update', 'replace', 'delete'] } } }
      ],
      { fullDocument: 'updateLookup' } // 获取更新后的完整文档，方便后续处理
    );

    // 监听UniqueCoupon集合的核心操作
    const uniqueCouponStream = db.collection('UniqueCoupon').watch(
      [
        { $match: { operationType: { $in: ['insert', 'update', 'replace', 'delete'] } } }
      ],
      { fullDocument: 'updateLookup' }
    );

    // 绑定事件处理函数
    attachStreamHandlers(couponRuleStream, 'CouponCartRule');
    attachStreamHandlers(uniqueCouponStream, 'UniqueCoupon');

    console.log('Change Streams初始化完成，开始监听集合更新');
  } catch (err) {
    console.error('初始化Change Streams失败:', err);
    // 失败后自动重试（指数退避更友好，这里简化为固定延迟）
    setTimeout(() => setupChangeStreams(), 5000);
  }
}

// 通用的流事件处理函数
function attachStreamHandlers(stream, collectionName) {
  // 处理更新事件
  stream.on('change', async (changeEvent) => {
    try {
      await processChangeEvent(collectionName, changeEvent);
    } catch (processErr) {
      console.error(`处理${collectionName}更新事件失败:`, processErr);
      // 可以将失败事件存入死信队列，后续人工排查或自动重试
      await saveToDeadLetterQueue(collectionName, changeEvent, processErr);
    }
  });

  // 处理流错误，自动重连
  stream.on('error', (err) => {
    console.error(`${collectionName} Change Stream出错:`, err);
    stream.close();
    setTimeout(() => setupChangeStreams(), 10000);
  });

  // 流关闭时自动重启
  stream.on('close', () => {
    console.log(`${collectionName} Change Stream已关闭，准备重启`);
    setTimeout(() => setupChangeStreams(), 5000);
  });
}

// 业务逻辑处理函数
async function processChangeEvent(collectionName, changeEvent) {
  // 1. 记录审计日志（持久化事件，保证可追溯）
  const auditDb = client.db('audit-db');
  const auditDoc = {
    collection: collectionName,
    operationType: changeEvent.operationType,
    documentId: changeEvent.documentKey._id,
    fullDocument: changeEvent.fullDocument,
    updateDetails: changeEvent.updateDescription, // 仅update操作包含字段变更信息
    eventTime: new Date(),
    resumeToken: changeEvent._id // 保存断点续传的token
  };
  await auditDb.collection('collection_update_logs').insertOne(auditDoc);

  // 2. 针对不同集合的业务逻辑处理
  if (collectionName === 'CouponCartRule') {
    // 示例：更新缓存中的优惠券规则
    // await redisClient.set(`coupon:rule:${changeEvent.documentKey._id}`, JSON.stringify(changeEvent.fullDocument));
    // 示例：如果maxUsage或maxBudget变更，触发关联优惠券的校验
    if (changeEvent.updateDescription?.updatedFields?.maxUsage || changeEvent.updateDescription?.updatedFields?.maxBudget) {
      await validateAssociatedCoupons(changeEvent.documentKey._id);
    }
  } else if (collectionName === 'UniqueCoupon') {
    // 示例：同步更新对应规则的currentUsage统计
    await updateRuleCurrentUsage(changeEvent.fullDocument.ruleId, changeEvent.fullDocument.currentCouponUsage);
  }
}

// 死信队列保存失败事件
async function saveToDeadLetterQueue(collectionName, event, error) {
  const auditDb = client.db('audit-db');
  await auditDb.collection('failed_update_events').insertOne({
    collection: collectionName,
    event: event,
    errorMessage: error.message,
    errorStack: error.stack,
    createdAt: new Date()
  });
}

// 关联优惠券校验（示例业务逻辑）
async function validateAssociatedCoupons(ruleId) {
  const db = client.db('your-business-db');
  const rule = await db.collection('CouponCartRule').findOne({ id: ruleId });
  const coupons = await db.collection('UniqueCoupon').find({ ruleId }).toArray();
  
  // 校验逻辑：比如检查优惠券总使用量是否超过规则的maxUsage
  const totalUsage = coupons.reduce((sum, coupon) => sum + coupon.currentCouponUsage, 0);
  if (totalUsage > rule.maxUsage) {
    // 触发预警：比如发送通知到运维系统
    console.warn(`规则${ruleId}下的优惠券总使用量已超过上限！`);
  }
}

// 更新规则的currentUsage（示例业务逻辑）
async function updateRuleCurrentUsage(ruleId, couponUsageChange) {
  const db = client.db('your-business-db');
  // 这里可以根据实际业务逻辑做累加或覆盖更新
  await db.collection('CouponCartRule').updateOne(
    { id: ruleId },
    { $inc: { currentUsage: couponUsageChange } }
  );
}

// 启动监听
setupChangeStreams();

• 断点续传：在审计日志中保存resumeToken，如果服务重启，可以通过watch({ resumeAfter: lastResumeToken })从上次中断的位置继续监听，完全不会丢失事件；
• 错误重试与死信队列：业务处理失败的事件会存入死信队列，支持后续人工排查或定时重试，避免数据不一致；
• 自动重连机制：流出现错误或关闭时自动重启监听，保证服务持续可用。

• 分布式消费：如果单节点监听压力大，可以将Change Streams的事件转发到Kafka/RabbitMQ等消息队列，多个消费者节点并行处理；
• 动态配置监听集合：通过配置中心管理需要监听的集合列表，无需修改代码即可新增/移除追踪目标；
• 分片集群适配：如果后续业务扩容到分片集群，Change Streams会自动跨分片监听，无需修改监听逻辑。

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