How to consume multiple jobs in bull at the same time?

Viewed 747

In some cases there is a relatively high amount of concurrency, but at the same time the importance of real-time is not high, so I am trying to use bull to create a queue.

Because the performance of the bulk request API will be significantly higher than the split to a single request, so I want to be able to consume multiple jobs in a function to call the bulk API at the same time

Here is my current code (with nestjs):

@Injectable()
export class DialogQueueService {
  constructor(
    @InjectQueue(QUEUE_NAME) public queue: Queue<DialogData>,
    private index: DialogsIndexService,
  ) {
  }

  async push(data: DialogData) {
    return this.queue.add(data, { jobId: data.id });
  }

  @Cron(CronExpression.EVERY_30_SECONDS)
  async process() {
    const jobs = (
      await this.queue.getWaiting(await this.queue.getWaitingCount())
    ).filter(Boolean);
    const total = jobs.length;
    try {
        const response = await this.index.bulkInsert(
          jobs.map((job) => job.data),
        );
        jobs.forEach((job) => job.moveToCompleted(null, true, true));
        return response;
      } catch (err) {
        jobs.forEach((job) => job.moveToFailed(err, true));
      }
  }
}

The current code has the following problems

  1. no queue events will be triggered
  2. the queue stored in Redis will be stuck at waiting state (even if the job itself has been deleted), which will cause the queue.getWaiting() function to block the event loop for a long time

Is there any elegant way to consume multiple jobs in bull at the same time?

0 Answers
Related