多消费者场景下单一生产者的选择方案咨询

不需要用复杂的集群选主机制，分布式锁就可以完美解决你的问题——这是周期性单实例任务场景下的标准方案，比集群选主更轻量、更贴合需求。

核心思路 #

利用Redis的原子命令实现跨实例的锁竞争：只有抢到锁的实例才执行任务，其他实例直接跳过。锁要设置合理的过期时间（略长于任务执行周期，比如70秒，避免实例意外挂掉导致锁无法释放）。

NodeJS实现方案 #

1. 基于Redis原生命令手动实现（轻量） #

用ioredis库配合SET命令的NX（仅当key不存在时设置）和PX（过期时间）参数实现原子锁，再用Lua脚本安全释放锁：

const Redis = require('ioredis');
const redis = new Redis({ /* Redis配置 */ });

async function executeExpiredTask() {
  const lockKey = 'expired-objects-producer-lock';
  const lockTtl = 70000; // 70秒，覆盖1分钟周期+任务执行时间
  const lockValue = `instance-${process.pid}`; // 用实例唯一标识作为锁值，确保仅自己能释放

  // 原子性获取锁
  const lockAcquired = await redis.set(lockKey, lockValue, 'NX', 'PX', lockTtl);
  
  if (!lockAcquired) {
    console.log('锁已被其他实例持有，跳过本次任务');
    return;
  }

  try {
    // 1. 查询数据库中过期的对象
    const expiredItems = await yourDBQueryFunction();
    // 2. 推送到消息队列
    await yourQueuePushFunction(expiredItems);
    console.log('过期对象任务执行完成');
  } finally {
    // 安全释放锁：仅当锁值匹配当前实例时才删除
    const releaseScript = `
      if redis.call('GET', KEYS[1]) == ARGV[1] then
        return redis.call('DEL', KEYS[1])
      else
        return 0
      end
    `;
    await redis.eval(releaseScript, 1, lockKey, lockValue);
  }
}

// 每分钟触发一次任务
setInterval(executeExpiredTask, 60000);

2. 使用成熟的分布式锁库（可靠） #

如果需要应对Redis集群或更高的可靠性要求，推荐使用实现Redlock算法的@node-redis/redlock库，它能在多Redis节点环境下保证锁的有效性：

const { createClient } = require('redis');
const Redlock = require('@node-redis/redlock');

// 初始化Redis客户端
const redisClient = createClient({ /* Redis配置 */ });
await redisClient.connect();

// 初始化Redlock实例
const redlock = new Redlock(
  [redisClient],
  {
    driftFactor: 0.01, // 允许的时钟偏差比例
    retryCount: 3,     // 获取锁失败后的重试次数
    retryDelay: 200,   // 重试间隔（毫秒）
    retryJitter: 200   // 重试随机抖动（毫秒）
  }
);

async function executeExpiredTask() {
  let lock;
  try {
    // 获取锁，有效期70秒
    lock = await redlock.lock('expired-objects-producer-lock', 70000);
    
    // 执行任务逻辑
    const expiredItems = await yourDBQueryFunction();
    await yourQueuePushFunction(expiredItems);
    console.log('过期对象任务执行完成');
  } catch (err) {
    console.log('无法获取锁，跳过任务:', err.message);
    return;
  } finally {
    // 释放锁（忽略释放失败的情况，锁会自动过期）
    if (lock) {
      await lock.unlock().catch(err => console.log('释放锁失败:', err.message));
    }
  }
}

// 每分钟触发一次任务
setInterval(executeExpiredTask, 60000);

关键注意点 #

• 锁的过期时间必须大于任务最长执行时间+周期间隔，防止任务未执行完锁就过期，导致多个实例同时执行。
• 释放锁时必须验证锁的归属（用唯一值匹配），避免释放其他实例持有的锁。
• 如果你的任务执行时间可能超过锁的TTL，可以考虑在任务执行过程中续期锁（比如用定时器定期延长锁的有效期）。

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