使用BullMQ拆分数据执行定时任务时如何解决重复执行问题？

我用Node.js的BullMQ模块开发定时任务，要把3000个用户拆成3个任务，每个处理1000个。但运行后任务大量重复执行，预期每个任务只跑一次，实际多次重复，怀疑是缓存任务导致。

相关代码 #

const Queue = require('bullmq').Queue;
const Worker = require('bullmq').Worker;
const IORedis = require('ioredis');
const connection = new IORedis({maxRetriesPerRequest: null});
  
const myQueue = new Queue('MyQueue', {connection});

myQueue.add('job1', {page: 1}, {
  repeat: {
      pattern: '* * * * *'
    },
  });
myQueue.add('job2', {page: 2}, {
  repeat: {
      pattern: '* * * * *'
    },
  });
myQueue.add('job3', {page: 3}, {
  repeat: {
      pattern: '* * * * *'
    },
  });

const worker = new Worker('MyQueue', async job => {
  if (['job1', 'job2', 'job3'].includes(job.name)) {
    console.log(`Processing ${job.name}`);
  } else {
    console.log(`Skipping ${job.name}`);
  }
}, {connection});

预期输出 #

Processing job1
Job1 with id repeat:0f6579d200bf5a419aff3c8e27c6844f:1713224040000 has been completed
Processing job2
Job2 with id repeat:31da86090fb08e946da917ae7d7803a3:1713224040000 has been completed
Processing job3
Job3 with id repeat:2fcf2b3f5b4c75f261e1b18122de6218:1713224040000 has been completed

实际输出 #

Processing job1
Job1 with id repeat:0f6579d200bf5a419aff3c8e27c6844f:1713224040000 has been completed
Processing job2
Job2 with id repeat:31da86090fb08e946da917ae7d7803a3:1713224040000 has been completed
Processing job3
Job3 with id repeat:2fcf2b3f5b4c75f261e1b18122de6218:1713224040000 has been completed
Processing job2
Job2 with id repeat:43460e979617272ff738de6b1bc91ac5:1713224040000 has been completed
Processing job3
Job3 with id repeat:6466183965f2fcfc37fb4a45eebf19ee:1713224040000 has been completed
Processing job2
Job2 with id repeat:d453511562d3dcf31136cfc737433a85:1713224040000 has been completed
Processing job3
Job3 with id repeat:ebe355420d8b30272e351063cc3073e2:1713224040000 has been completed
Processing job2
Job2 with id repeat:63d53b94e1f84e3ad2b2507a3f59ecc0:1713224040000 has been completed
Processing job3
Job3 with id repeat:a07aabec19f83a426707551ef36e9659:1713224040000 has been completed
Processing job2
Job2 with id repeat:bf4162a60feabd3c853e15a853046382:1713224040000 has been completed
Processing job3
Job3 with id repeat:db44ec844a46f4f7209f129667c32f05:1713224040000 has been completed
...

问题原因及解决方法 #

核心原因 #

1. 重复添加定时任务：每次启动应用时，代码都会执行myQueue.add()三次，向队列中重复添加相同名称的定时任务。BullMQ会将这些视为独立的重复任务，导致同一时间点出现多个同名任务实例。
2. Cron表达式设置：你用的* * * * *表示每分钟执行一次，这本身就会让任务每分钟重复运行，和你“每个任务仅执行一次”的预期不符。

解决步骤 #

1. 避免重复添加任务：在添加任务前，先检查队列中是否已存在同名的重复任务，只有不存在时才添加：

async function addJobsIfNotExists() {
  const existingJobs = await myQueue.getRepeatableJobs();
  const existingJobNames = existingJobs.map(job => job.name);
  
  if (!existingJobNames.includes('job1')) {
    await myQueue.add('job1', {page: 1}, {
      repeat: { pattern: '* * * * *' } // 如果仅需执行一次，直接移除repeat配置
    });
  }
  if (!existingJobNames.includes('job2')) {
    await myQueue.add('job2', {page: 2}, {
      repeat: { pattern: '* * * * *' }
    });
  }
  if (!existingJobNames.includes('job3')) {
    await myQueue.add('job3', {page: 3}, {
      repeat: { pattern: '* * * * *' }
    });
  }
}

addJobsIfNotExists();

2. 调整任务执行策略：
   • 如果任务只需执行一次：直接移除repeat配置，任务添加后只会运行一次。
   • 如果需要定时重复执行：确保只添加一次重复任务，避免应用重启时重复添加。
3. 清理已有重复任务：先手动清理Redis中已存在的重复任务，避免残留的旧任务继续执行：

async function cleanDuplicateJobs() {
  const existingJobs = await myQueue.getRepeatableJobs();
  for (const job of existingJobs) {
    if (['job1', 'job2', 'job3'].includes(job.name)) {
      await myQueue.removeRepeatableByKey(job.key);
    }
  }
}

// 先清理再添加新任务
cleanDuplicateJobs().then(addJobsIfNotExists);

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