BullMQ任务重复执行问题求助（附Node.js实现代码）

核心原因 #

你遇到的任务重复执行及相关错误，本质是分布式锁失效或Redis通信异常导致的：GCP App Engine标准环境的实例会自动启停/缩容，加上Redis网络波动，可能造成Worker的任务锁未正确续约、释放，进而导致其他Worker重新拾取已在执行的任务。

针对性解决方案 #

1. 优化Worker锁配置 #

调整BullMQ Worker的锁超时和续约参数，确保任务处理期间锁不会意外过期：

const worker = new Worker('task-queue', processor, {
  connection: redisConfig,
  // 根据任务平均处理时间调整，默认15秒，延长至30秒
  lockDuration: 30000,
  // 每10秒续约一次锁，降低网络波动导致续约失败的概率
  lockRenewTime: 10000,
  // 设置锁的最大存活时间，防止实例重启后残留锁阻塞任务
  maxLockDuration: 60000
});

2. 加固Redis可靠性 #

• 确保GCP Redis Memorystore开启持久化（RDB/AOF），避免Redis重启丢失任务锁键；
• 使用与App Engine同区域的Redis实例，降低跨区域网络延迟；
• 监控Redis的连接数和慢查询日志，排查是否存在Redis操作超时导致锁设置失败的情况。

3. 实现任务幂等性（兜底方案） #

分布式环境下锁机制无法100%避免极端情况，必须给业务逻辑加幂等控制：

• 提交任务时指定唯一jobId，避免重复入队：

  queue.add('task-name', taskData, { jobId: `task-${userId}-${timestamp}` });
  

• 业务函数执行前，先查询数据库/缓存中该任务的执行状态，若已完成直接返回；执行完成后标记任务为已处理。

4. 完善Worker异常处理 #

• 用try/catch包裹处理器的所有逻辑，确保异步错误被捕获，避免Worker进程崩溃导致锁未释放：

  const processor = async (job) => {
    try {
      await businessFunction(job.data);
    } catch (err) {
      console.error(`Job ${job.id} failed:`, err);
      // 不要手动修改任务状态，交给BullMQ自动处理
      throw err;
    }
  };
  

• 监听Worker的failed事件，排查任务失败的具体原因：

  worker.on('failed', (job, err) => {
    console.error(`Job ${job.id} failed with error:`, err.message);
  });
  

5. 调整App Engine实例配置 #

• 限制自动缩放的实例数量上限，避免短时间内实例暴增导致Redis连接过载；
• 给实例分配足够的CPU和内存，防止资源不足导致Worker卡顿、锁续约不及时；
• 关闭App Engine的自动休眠（若启用），避免实例休眠期间锁无法续约。

错误对应原因说明 #

• Missing key for job "abc"：Redis中任务的元数据键被意外删除，可能是Redis重启、持久化失败或Worker异常退出未清理；
• Missing lock for job "abc"：任务锁已过期，其他Worker已拾取该任务；
• Job "abc" is not in the active state：当前Worker尝试操作已被其他Worker标记为完成/失败的任务；
• Lock mismatch for job "abc"：Worker持有的锁与Redis中的锁不一致，通常是锁续约失败后被其他Worker重新获取。

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