如何使用BullMQ处理cronjob的块数据？

Question

我正在使用 BullMQ 节点模块开发 cronjob。 我有 3000 个用户，我想分离 3 个作业，其中一个作业用于 1000 个用户，并处理 cronjob。 但是当我运行这段代码时，它会执行许多任务。

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
...

很多工作都在进行。 这是缓存的作业吗？

Answer 1

如果您连接到 Redis 并运行

zrange bull:MyQueue:repeat 0 100

，您应该会看到 MyQueue 每分钟重复一次的作业列表。如果参数完全相同，

bullmq

足够聪明，当您运行

myQueue.add('job1'..

时不会重复 cron 条目，但听起来您已经缓存了 cron 作业，这些作业在您添加工作程序时也在运行。

我在本地运行了您的代码，它确实按照您想要的方式工作：它启动了一个每分钟处理一次作业的工作人员。多次启动worker并不会创建多个可重复的作业，所以我认为这与存储在redis中的数据有关。

zrange bull:MyQueue:repeat 0 100
1) "job1::::* * * * *"
2) "job2::::* * * * *"
3) "job3::::* * * * *"