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
- no queue events will be triggered
- the queue stored in Redis will be stuck at
waitingstate (even if the job itself has been deleted), which will cause thequeue.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?