Node.js In-Depth
Module P-12·18 min read

Why job queues exist, BullMQ with Redis, workers and concurrency, retries and exponential backoff, dead letter queues, cron jobs, and real-world use cases.

Written by Jatin Jain Saraf · Architect of SupraScan

Module P-12 — Background Jobs and Task Queues with BullMQ

What this module covers: Some work is too slow, too unreliable, or too risky to do inside a request. Sending email, resizing images, calling third-party APIs, generating reports, sending push notifications — all of these should happen off the request path. This module covers why job queues exist, BullMQ's architecture with Redis, defining and dispatching jobs, writing workers with concurrency control, retries with exponential backoff, dead letter queues for failed jobs, repeatable cron jobs, and the patterns that make queues reliable in production.

Why Job Queues

A job queue is a restaurant's order rail — the waiter can hand off a ticket the instant it's written and get back to the floor, while the kitchen fires it at its own pace.

Consider sending a transactional email when an order is placed. Three approaches:

// Option 1: Inline — user waits 300ms for Mailgun
export const createOrder = asyncHandler(async (req, res) => {
  const order = await ordersService.createOrder(req.body);
  await emailService.sendConfirmation(order);   // blocks the response
  res.status(201).json(order);
});
 
// Option 2: Fire-and-forget — fast but unreliable
export const createOrder = asyncHandler(async (req, res) => {
  const order = await ordersService.createOrder(req.body);
  emailService.sendConfirmation(order).catch(console.error); // if this fails, no retry
  res.status(201).json(order);
});
 
// Option 3: Job queue — fast AND reliable
export const createOrder = asyncHandler(async (req, res) => {
  const order = await ordersService.createOrder(req.body);
  await emailQueue.add('send-confirmation', { orderId: order.id }); // returns in <1ms
  res.status(201).json(order); // response goes out immediately
  // Worker picks up the job, retries on failure, logs results
});

A job queue gives you:

  • Decoupled execution — the response returns before the work is done
  • Automatic retries — if the email provider is down, the job retries with backoff
  • Visibility — you can see queued, active, completed, and failed jobs
  • Rate limiting — process at most N jobs per second regardless of how many are enqueued
  • Concurrency control — run M workers in parallel without overloading the system
  • Persistence — jobs survive process restarts (stored in Redis)

BullMQ Architecture

BullMQ uses Redis as its store. Three components:

Producer (your API)     Queue (Redis)        Worker (separate process or same process)
       │                     │                           │
       │── queue.add() ──────►│                           │
       │                     │◄──── worker.process() ────│
       │                     │                           │
       │                     │──── job data ────────────►│
       │                     │                           │ (processes job)
       │                     │◄──── completed ───────────│

The queue lives in Redis — producers and workers are just Node.js processes that connect to it. You can have multiple producers (different API instances) and multiple workers (separate scaling).

npm install bullmq ioredis

Defining Queues and Adding Jobs

// src/queues/email.queue.ts
import { Queue } from 'bullmq';
import IORedis from 'ioredis';
import { env } from '../config/env.js';
 
// Job type definitions — TypeScript safety across producer and worker
export interface SendOrderConfirmationJob {
  orderId: number;
}
 
export interface SendPasswordResetJob {
  userId: number;
  resetToken: string;
}
 
export type EmailJobData =
  | { type: 'order-confirmation'; data: SendOrderConfirmationJob }
  | { type: 'password-reset'; data: SendPasswordResetJob };
 
// Create the queue — connects to Redis
export const emailQueue = new Queue<EmailJobData>('email', {
  connection: new IORedis(env.REDIS_URL, { maxRetriesPerRequest: null }),
  defaultJobOptions: {
    attempts: 3,                        // retry up to 3 times
    backoff: {
      type: 'exponential',
      delay: 1000,                      // 1s, 2s, 4s
    },
    removeOnComplete: { count: 1000 },  // keep last 1000 completed jobs
    removeOnFail: { count: 5000 },      // keep last 5000 failed jobs for debugging
  },
});

Adding jobs from your service:

// src/services/orders.service.ts
import { emailQueue } from '../queues/email.queue.js';
 
export async function createOrder(input: CreateOrderInput) {
  const order = await ordersRepo.create(input);
 
  // Add job — returns in < 1ms
  await emailQueue.add('send-confirmation', {
    type: 'order-confirmation',
    data: { orderId: order.id },
  });
 
  return order;
}

Job options per-add

// High priority — processed before normal jobs
await notificationQueue.add('push', payload, { priority: 1 });
 
// Delay — run 10 minutes from now (e.g. "your session expires soon")
await emailQueue.add('expiry-warning', payload, { delay: 10 * 60 * 1000 });
 
// Unique job — skip if a job with this key already exists
await reportQueue.add('daily-report', payload, {
  jobId: `daily-report:${today}`,  // deterministic ID prevents duplicates
});
 
// LIFO — process newest first (useful for real-time notifications)
await notificationQueue.add('push', payload, { lifo: true });

Writing Workers

A worker subscribes to a queue and processes jobs:

// src/workers/email.worker.ts
import { Worker, Job } from 'bullmq';
import IORedis from 'ioredis';
import { EmailJobData } from '../queues/email.queue.js';
import * as emailService from '../services/email.service.js';
import * as ordersRepo from '../repositories/orders.repository.js';
import * as usersRepo from '../repositories/users.repository.js';
import logger from '../utils/logger.js';
import { env } from '../config/env.js';
 
const worker = new Worker<EmailJobData>(
  'email',
  async (job: Job<EmailJobData>) => {
    logger.info({ jobId: job.id, type: job.data.type }, 'Processing email job');
 
    switch (job.data.type) {
      case 'order-confirmation': {
        const { orderId } = job.data.data;
        const order = await ordersRepo.findById(orderId);
        if (!order) throw new Error(`Order ${orderId} not found`);
 
        const user = await usersRepo.findById(order.userId);
        if (!user) throw new Error(`User ${order.userId} not found`);
 
        await emailService.sendOrderConfirmation({
          to: user.email,
          orderNumber: order.id.toString(),
          items: order.items,
          total: order.total,
        });
        break;
      }
 
      case 'password-reset': {
        const { userId, resetToken } = job.data.data;
        const user = await usersRepo.findById(userId);
        if (!user) throw new Error(`User ${userId} not found`);
 
        await emailService.sendPasswordReset(user.email, resetToken);
        break;
      }
 
      default:
        throw new Error(`Unknown job type: ${(job.data as any).type}`);
    }
 
    logger.info({ jobId: job.id }, 'Email job completed');
  },
  {
    connection: new IORedis(env.REDIS_URL, { maxRetriesPerRequest: null }),
    concurrency: 5,        // process up to 5 jobs simultaneously
    limiter: {
      max: 10,             // max 10 jobs per...
      duration: 1000,      // ...1 second (rate limit)
    },
  },
);
 
// Event listeners for observability
worker.on('completed', (job) => {
  logger.info({ jobId: job.id, type: job.data.type }, 'Job completed');
});
 
worker.on('failed', (job, err) => {
  logger.error({ jobId: job?.id, type: job?.data.type, err }, 'Job failed');
});
 
worker.on('error', (err) => {
  logger.error({ err }, 'Worker error');
});
 
export default worker;

Sign in to keep reading

The rest of this module is free — sign in with Google to unlock it and track your progress.

Sign in & Register
PreviousModule P-11: GraphQL with Apollo ServerNext Module P-13: JSON Internals, Serialization, and Schema Validation

Discussion

0

Join the discussion

Loading comments...
PortfolioBlog

© 2026 Jatin Jain Saraf (JJS). All rights reserved.