Queues with BullMQ

Why Queues

Sending an email, generating an invoice PDF or calling a slow courier API should not make the customer wait. Put a job in a queue and return immediately; a worker processes it in the background, with retries if it fails. BullMQ stores jobs in Redis.

Example

npm i @nestjs/bullmq bullmq

Example output

added 12 packages in 4s

src/emails/emails.module.ts

import { BullModule } from '@nestjs/bullmq';
import { Module } from '@nestjs/common';
import { EmailsProcessor } from './emails.processor';
import { EmailsService } from './emails.service';

@Module({
  imports: [BullModule.registerQueue({ name: 'emails' })],
  providers: [EmailsService, EmailsProcessor],
  exports: [EmailsService],
})
export class EmailsModule {}

// app.module.ts: BullModule.forRoot({ connection: { host: 'localhost', port: 6379 } })

Producer and Worker

src/emails/emails.service.ts (producer)

import { InjectQueue } from '@nestjs/bullmq';
import { Injectable } from '@nestjs/common';
import { Queue } from 'bullmq';

@Injectable()
export class EmailsService {
  constructor(@InjectQueue('emails') private readonly queue: Queue) {}

  orderConfirmation(orderId: string, email: string) {
    return this.queue.add(
      'order-confirmation',
      { orderId, email },
      { attempts: 5, backoff: { type: 'exponential', delay: 2000 }, removeOnComplete: 1000, removeOnFail: 5000 },
    );
  }
}

src/emails/emails.processor.ts (worker)

import { OnWorkerEvent, Processor, WorkerHost } from '@nestjs/bullmq';
import { Logger } from '@nestjs/common';
import { Job } from 'bullmq';
import { Mailer } from './mailer';

@Processor('emails', { concurrency: 10 })
export class EmailsProcessor extends WorkerHost {
  private readonly logger = new Logger(EmailsProcessor.name);

  constructor(private readonly mailer: Mailer) {
    super();
  }

  async process(job: Job<{ orderId: string; email: string }>): Promise<void> {
    switch (job.name) {
      case 'order-confirmation':
        await this.mailer.send(job.data.email, 'Your ShopEase order is confirmed', { orderId: job.data.orderId });
        return;
      default:
        throw new Error(`Unknown job ${job.name}`);
    }
  }

  @OnWorkerEvent('failed')
  onFailed(job: Job, err: Error) {
    this.logger.warn(`job ${job.id} (${job.name}) attempt ${job.attemptsMade} failed: ${err.message}`);
  }
}

Example

npm run start:dev

Example output

[Nest] 25010  - WARN [EmailsProcessor] job 412 (order-confirmation) attempt 1 failed: connect ETIMEDOUT smtp.mailgun.org:587
[Nest] 25010  - WARN [EmailsProcessor] job 412 (order-confirmation) attempt 2 failed: connect ETIMEDOUT smtp.mailgun.org:587
[Nest] 25010  - LOG [Mailer] sent "Your ShopEase order is confirmed" to [email protected]

Retries with Exponential Backoff

Example

With attempts: 5 and exponential backoff of 2 s, BullMQ waits 2 s, 4 s, 8 s, 16 s between attempts. A simulation of a job that succeeds on the third try (delays scaled down to milliseconds):

const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms));

async function runJob(name: string, work: (attempt: number) => Promise<string>, attempts: number, baseDelay: number) {
  for (let attempt = 1; attempt <= attempts; attempt++) {
    try {
      const result = await work(attempt);
      console.log(name, 'attempt', attempt, 'ok:', result);
      return;
    } catch (e) {
      const wait = baseDelay * 2 ** (attempt - 1);
      console.log(name, 'attempt', attempt, 'failed:', (e as Error).message, attempt < attempts ? '-> retry in ' + wait + 'ms' : '-> moved to failed');
      if (attempt < attempts) await sleep(wait);
    }
  }
}

async function main() {
  await runJob('email#412', async (n) => {
    if (n < 3) throw new Error('SMTP timeout');
    return 'sent to [email protected]';
  }, 5, 2);
  await runJob('sms#413', async () => { throw new Error('invalid number'); }, 3, 2);
}
main();

Output

email#412 attempt 1 failed: SMTP timeout -> retry in 2ms
email#412 attempt 2 failed: SMTP timeout -> retry in 4ms
email#412 attempt 3 ok: sent to [email protected]
sms#413 attempt 1 failed: invalid number -> retry in 2ms
sms#413 attempt 2 failed: invalid number -> retry in 4ms
sms#413 attempt 3 failed: invalid number -> moved to failed