Node.js多实例下SQS FIFO队列消费者序列化实现方案咨询

现有方案核心问题 #

你的初始MongoDB锁方案存在饥饿风险：当锁被释放后，所有实例同时抢锁，同一个实例可能连续两次抢到，破坏任务按入队顺序执行的要求；而序列计数器方案则存在单点故障风险，若持有当前序列的实例崩溃，会导致后续任务全部停滞。

最优方案：利用SQS FIFO队列原生特性 #

如果你的业务场景允许切换队列类型，直接使用SQS FIFO队列是最省心、可靠的方案：

• SQS FIFO队列默认保证消息按入队顺序处理，且同一MessageGroupId下的消息会被逐个分发（同一时间只有一个实例处理该Group的消息）。
• 若所有任务需要严格全局顺序，只需给所有消息设置相同的MessageGroupId，SQS会自动确保同一时间只有一个实例处理消息，处理完成后才会分发下一条，完全不需要额外的锁机制。

基于现有MongoDB的改进方案（避免饥饿） #

如果无法切换到FIFO队列，可以修改MongoDB锁的抢锁逻辑，实现公平锁避免饥饿：

1. 给锁文档增加lastOwner字段，记录上一个持有锁的实例ID
2. 抢锁时增加条件：当前锁未被持有，且上一个持有者不是当前实例，确保同一实例不会连续抢锁
3. 释放锁时更新lastOwner为当前实例，而不是直接删除锁（避免频繁创建/删除文档，同时保留历史持有者信息）

改进后的代码示例 #

const LockSchema = new Schema({
  id: String,
  owner: String, // 当前持有锁的实例ID
  lastOwner: String, // 上一个持有锁的实例ID，用于避免饥饿
  createdAt: { type: Date, required: true, default: Date.now },
});

// 设置锁的过期时间，防止实例崩溃导致锁永久占用
LockSchema.index({ createdAt: 1 }, { expireAfterSeconds: 600 });

const Lock = mongoose.model("queue-lock", LockSchema, "locks");

// 监听锁的变更事件（释放或更新）
const filter = [
  {
    $match: {
      $or: [{ operationType: "update" }, { operationType: "insert" }]
    }
  }
];

const options = { fullDocument: 'updateLookup' };
const lockStream = Lock.watch(filter, options);

// 处理锁变更事件
lockStream.on('change', async (event) => {
  // 锁已被释放（owner为空）
  if (!event.fullDocument?.owner) {
    await attemptLock();
  }
});

// 抢锁逻辑
async function attemptLock() {
  const result = await Lock.findOneAndUpdate(
    { 
      id: "migration-lock",
      owner: { $eq: null }, // 锁未被持有
      lastOwner: { $ne: whoAmI } // 上一个持有者不是当前实例，避免连续抢锁
    },
    { 
      $set: { owner: whoAmI, createdAt: new Date() },
      $setOnInsert: { lastOwner: null }
    },
    { upsert: true, new: true }
  ).lean();

  if (result.owner === whoAmI) {
    // 成功获取锁，处理任务
    try {
      await processNextSQSTask(); // 替换为你的SQS任务处理逻辑
    } finally {
      // 释放锁，更新lastOwner
      await Lock.findOneAndUpdate(
        { id: "migration-lock" },
        { $set: { owner: null, lastOwner: whoAmI } }
      );
    }
  }
}

// 实例启动时尝试第一次抢锁
(async () => {
  await attemptLock();
})();

备选方案：Redis公平队列锁 #

如果你的系统已经使用Redis，也可以用Redis实现公平锁：

• 用Redis List维护等待队列，实例抢锁前先将自己的ID推入队列
• 只有当队列头部的ID是当前实例ID时，才能获取锁处理任务
• 处理完成后将自己的ID从队列头部移除，确保下一个实例可以抢锁
  Redis的原子操作（如LPUSH、LSET、BLPOP）可以完美实现这种公平队列，性能比MongoDB更适合高频锁场景。

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