BullMQ按指定ID获取队列任务后调用moveToCompleted报缺锁错误如何解决

问题原因 #

你触发Missing lock for job报错的核心原因是：BullMQ的分布式锁机制要求，只有持有job有效锁的进程才能调用moveToCompleted这类修改job生命周期状态的接口，你直接通过getJob获取的job实例没有提前抢占锁，因此操作被拦截。

解决方案 #

你只需在调用moveToCompleted之前，先对目标job加锁即可，修改后的代码如下：

const jobID = "THE_ID_OF_THE_JOB_I_AM_WAITING_FOR";
const token = `${jobID}_results_worker`;
const queueEvents = new QueueEvents('results');
const resQueue = this.queues.get('results');
// 锁超时时间，单位毫秒，需大于你处理业务的最大耗时
const LOCK_TTL = 30 * 1000;

const waitResult = async (job: {jobId: any}) => {
  if (job.jobId === jobID){
    debug(`Result job for ${jobID} received!`);
    const resJob = await resQueue?.getJob(jobID) as Job;
    queueEvents.removeListener('waiting', waitResult);
    try {
      // 先抢占目标job的锁
      await resJob.lock(token, LOCK_TTL);
      // 加锁成功后再更新状态
      await resJob.moveToCompleted('Results received', token, false);
      resolve(resJob?.data);
    } catch (err) {
      // 锁抢占失败一般是有其他进程已经在处理该job，可根据业务需要做重试或异常处理
      reject(err);
    }
  }
} 

const listener = queueEvents?.on('waiting', waitResult );

额外注意事项 #

• 如果你的场景不需要留存结果队列的历史任务数据，也可以在拿到resJob.data后直接调用await resJob.remove()删除任务，该操作不需要持有锁，实现更简单。
• 如果存在多个进程同时监听同一个jobID的情况，只有第一个抢到锁的进程能成功修改状态，建议根据业务逻辑保证同一个jobID仅被一个进程监听，或者添加合理的重试机制。

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