BullMQ

BullMQ is the modern, TypeScript-first successor to the original Bull library. It provides advanced features like Job Flows (DAGs), optimized Redis memory usage, and highly scalable worker management.

Overview

While the original @nestjs/bull package is still widely used, new NestJS projects should use @nestjs/bullmq.

BullMQ is essentially Bull v2. It was rewritten from the ground up to solve fundamental architectural limitations of the original library, specifically around how workers scale and how jobs can be linked together into complex workflows.

Key Concepts

• Strict Separation: In standard Bull, a Queue object could both add jobs and process jobs. In BullMQ, these are strictly separated into Queue (Producer) and Worker (Consumer) classes for better memory management.
• Job Flows: BullMQ allows you to create Parent-Child job relationships (Directed Acyclic Graphs). A parent job (render_video) will automatically wait in the queue until all child jobs (process_audio, process_frames) have successfully completed.
• Worker Concurrency: BullMQ workers are heavily optimized to run hundreds of asynchronous jobs concurrently without blocking the Node event loop.

Code Examples

1. Installation

Install the BullMQ package. (Note: BullMQ requires Redis v5.0.0 or higher!).
npm i @nestjs/bullmq bullmq

2. Global Configuration

Configuration in AppModule is almost identical to standard Bull.

// app.module.ts
import { Module } from '@nestjs/common';
import { BullModule } from '@nestjs/bullmq';

@Module({
  imports: [
    BullModule.forRoot({
      connection: {
        host: 'localhost',
        port: 6379,
      },
    }),
    // Register specific queues
    BullModule.registerQueue({
      name: 'video_processing',
    }),
  ],
})
export class AppModule {}

3. Producing Jobs (Queue)

Injecting and adding jobs works exactly the same, but notice you import Queue from bullmq, not bull.

// video.controller.ts
import { Controller, Post } from '@nestjs/common';
import { InjectQueue } from '@nestjs/bullmq';
import { Queue } from 'bullmq'; // MUST import from bullmq!

@Controller('video')
export class VideoController {
  constructor(@InjectQueue('video_processing') private videoQueue: Queue) {}

  @Post()
  async renderVideo() {
    await this.videoQueue.add('render_job', { videoId: 123 });
    return 'Processing...';
  }
}

4. Consuming Jobs (Worker)

In BullMQ, workers are defined using the @Processor() decorator (just like standard Bull), but the inner working is different. You typically extend the WorkerHost class.

// video.processor.ts
import { Processor, WorkerHost } from '@nestjs/bullmq';
import { Job } from 'bullmq';

// Tell NestJS this class is a BullMQ Worker for the 'video_processing' queue
@Processor('video_processing')
export class VideoProcessor extends WorkerHost {
  
  // You MUST implement the process() method. 
  // It acts as the catch-all for any job added to this queue.
  async process(job: Job<any, any, string>): Promise<any> {
    
    // Switch on the job name to handle different job types
    switch (job.name) {
      case 'render_job':
        console.log(`Rendering video ${job.data.videoId}...`);
        await this.renderLogic(job.data.videoId);
        return { success: true };
      default:
        throw new Error(`Unknown job name: ${job.name}`);
    }
  }

  private async renderLogic(id: number) {
    // Heavy lifting...
  }
}

Best Practices

• Use BullMQ FlowProducer for Complex Workflows: If you need to download a video, extract the audio, transcribe the audio, and then merge the subtitles back onto the video, do NOT try to orchestrate this manually. Use BullMQ’s FlowProducer. It allows you to declare a tree of jobs, and BullMQ natively ensures the root job only executes after all leaf jobs complete.
• Worker Sandboxing: Just like standard Bull, if your worker runs heavy CPU tasks (like video encoding using ffmpeg), you must sandbox the worker so it runs in a separate process. Otherwise, your main HTTP server will become completely unresponsive while the video renders.