Processors and Concurrency

A processor is a class whose process() method BullMQ calls once per job; concurrency decides how many it runs at a time.

Syntax@Processor(queueName: string, workerOptions?: NestWorkerOptions)

A processor is the consumer half of a queue: a class whose method BullMQ calls once per job. You declare it with @Processor('email'), extend WorkerHost, and implement process(job). The decorator only leaves metadata behind — you still list the class in its module's providers, which is how Nest's explorer finds it and, when the module starts, builds a BullMQ Worker around it. The worker connects to Redis, takes a job, locks it, and awaits your method. Return a value and the job is completed with that value stored on it. Throw and the job is failed.

Where this runs

Nowhere near a request. None of the HTTP pipeline applies — no middleware, guards, interceptors, pipes or exception filters. Validation and error mapping that a controller gets for free you do by hand here. Injected providers behave exactly as they do elsewhere.

Concurrency

A worker processes one job at a time by default (concurrency: 1). Raise it through the decorator's second argument and the worker keeps that many jobs in flight. That only helps while your code is awaiting I/O: two awaits interleave, two CPU-bound loops do not. CPU work in the API process also blocks the event loop, so HTTP requests queue behind image resizing. The fix is not a smaller number, it is a second process that runs only workers.

One class per queue

BullMQ has no per-job-name method decorator, so a single process() handles every name on that queue. Switch on job.name and throw on an unknown one.

Example

Example · typescript
// src/email/email.processor.ts
import { OnWorkerEvent, Processor, WorkerHost } from '@nestjs/bullmq';
import { Logger } from '@nestjs/common';
import { Job } from 'bullmq';
import { MailerService } from './mailer.service';

// Second argument is NestWorkerOptions: BullMQ's WorkerOptions minus
// "connection", which BullModule.forRoot() already supplied.
@Processor('email', { concurrency: 5 })
export class EmailProcessor extends WorkerHost {
  private readonly logger = new Logger(EmailProcessor.name);

  // WorkerHost has no constructor arguments, but you must still call super().
  constructor(private readonly mailer: MailerService) {
    super();
  }

  // Called once per job. The optional second parameter is the lock token,
  // needed only if you move the job yourself (moveToDelayed, moveToFailed).
  async process(job: Job<{ userId: string }, void, string>): Promise<void> {
    switch (job.name) {
      case 'welcome':
        await this.mailer.sendWelcome(job.data.userId);
        return;
      case 'weekly-digest':
        await this.mailer.sendDigest(job.data.userId);
        return;
      default:
        // A name nobody handles would otherwise complete silently.
        throw new Error(`Unknown job name on email queue: ${job.name}`);
    }
  }

  // Worker events are only read off the @Processor class itself.
  @OnWorkerEvent('completed')
  onCompleted(job: Job): void {
    this.logger.log(`${job.name} ${job.id} completed`);
  }

  // The job can be undefined: the worker may know a job failed without
  // having been able to load it.
  @OnWorkerEvent('failed')
  onFailed(job: Job | undefined, error: Error): void {
    this.logger.error(`${job?.name} ${job?.id} failed: ${error.message}`);
  }
}

When to use it

  • Resize uploaded images in a worker process so HTTP latency stays flat.
  • Hold a vendor integration to 10 calls per second with limiter, across every replica.
  • Run four S3 uploads at once per worker instead of one, since they are all awaits.
  • Expose per-job progress to an admin page during a long export.

More examples

Workers in their own process, so they cannot starve the API

Two entry points over one codebase: main.ts serves HTTP and only adds jobs, worker.ts only consumes them. Reach for this the first time a job does real CPU work, or when you want to scale workers separately from the API. The image is the same, the command differs: node dist/main.js against node dist/worker.js.

Example · typescript
// src/worker.ts
import { NestFactory } from '@nestjs/core';
import { WorkerModule } from './worker.module';

async function bootstrap(): Promise<void> {
  // createApplicationContext starts the DI container with no HTTP server.
  // Providers, @Processor classes and lifecycle hooks all still work.
  const app = await NestFactory.createApplicationContext(WorkerModule);

  // On SIGTERM, Nest runs onModuleDestroy/onApplicationShutdown, which is how
  // @nestjs/bullmq closes its workers: the current job finishes first.
  app.enableShutdownHooks();
}

void bootstrap();

// src/worker.module.ts
import { Module } from '@nestjs/common';
import { BullModule } from '@nestjs/bullmq';
import { EmailModule } from './email/email.module';
import { PdfModule } from './pdf/pdf.module';

@Module({
  imports: [
    BullModule.forRoot({
      connection: { host: process.env.REDIS_HOST ?? '127.0.0.1', port: 6379 },
    }),
    // Only the modules whose processors this process should run.
    EmailModule,
    PdfModule,
  ],
})
export class WorkerModule {}

Rate limiting a worker and reporting progress

concurrency caps what this process runs at once; limiter caps the rate across every worker on the queue, which is what a third-party quota actually needs. updateProgress is how a long job becomes visible to an admin screen, and the returned object is the job's result for anyone who later loads it by id.

Example · typescript
// src/pdf/pdf.processor.ts
import { Processor, WorkerHost } from '@nestjs/bullmq';
import { Job } from 'bullmq';
import { RendererService } from './renderer.service';

@Processor('pdf', {
  concurrency: 4, // four jobs in flight in this process
  // ...but no more than 10 jobs started per second, counted in Redis and so
  // shared by every worker on this queue. Use it to respect a vendor's quota.
  limiter: { max: 10, duration: 1000 },
})
export class PdfProcessor extends WorkerHost {
  constructor(private readonly renderer: RendererService) {
    super();
  }

  async process(job: Job<{ invoiceIds: string[] }>): Promise<{ urls: string[] }> {
    const urls: string[] = [];

    for (const [index, id] of job.data.invoiceIds.entries()) {
      urls.push(await this.renderer.render(id));
      // Writes to Redis and emits "progress"; a dashboard can poll job.progress.
      await job.updateProgress(
        Math.round(((index + 1) / job.data.invoiceIds.length) * 100),
      );
    }

    // The returned value is stored as job.returnvalue.
    return { urls };
  }
}

The two forms of @Processor

@Processor('x', opts) desugars to @Processor({ name: 'x' }, opts), so recognise both. The object form is the only way to pass configKey (a second Redis connection) or scope (a new processor instance per job). autorun: false leaves the worker built but idle until you call this.worker.run(), which is useful when the process must warm a cache first.

Example · typescript
// src/reports/reports.processor.ts
import { Processor, WorkerHost } from '@nestjs/bullmq';
import { Job } from 'bullmq';

// Shorthand: queue name first, worker options second.
@Processor('reports', { concurrency: 2 })
export class ReportsProcessor extends WorkerHost {
  async process(job: Job): Promise<void> {
    // ...
  }
}

// src/reports/reports-archive.processor.ts
import { Processor, WorkerHost } from '@nestjs/bullmq';
import { Job } from 'bullmq';

// Longhand: the same thing, with the Nest-specific options spelled out.
// configKey selects a named configuration registered with
// BullModule.forRoot('analytics', { ... }) - a second Redis, for example.
@Processor(
  { name: 'reports-archive', configKey: 'analytics' },
  { concurrency: 2, autorun: false }, // autorun: false means you start it yourself
)
export class ReportsArchiveProcessor extends WorkerHost {
  async process(job: Job): Promise<void> {
    // ...
  }
}

Related

nest queues bullmqnest queue retriesnest queue idempotencynest shutdownnest logging

Discussion

  • Be the first to comment on this lesson.