← Articles

Building a job queue with BullMQ and Redis

By Layla Hassan · 16 July 20260 views
#bullmq#redis#queues#node#typescript#workers#background-jobs#concurrency

Background jobs are the backbone of any serious backend system. Sending emails, generating PDFs, processing uploads, syncing external APIs, running scheduled reports — none of these belong in a request/response cycle. They should be queued, executed asynchronously, and retried on failure with observable progress.

BullMQ is the de-facto standard for this in the Node.js ecosystem. It is the spiritual successor to Bull, rewritten in TypeScript with first-class support for job priorities, rate limiting, flow producers (parent-child job trees), and a clean separation between the producer and worker roles. It stores everything in Redis, which means it is durable, supports clustering, and can handle millions of jobs without breaking a sweat.

This guide builds a complete job queue system from scratch — from a single worker to a production setup with concurrency tuning, failure recovery, repeatable jobs, and an operational dashboard.

Why BullMQ over plain Redis pub/sub or a database queue

It is worth addressing the alternatives before reaching for BullMQ. Redis pub/sub is fire-and-forget: messages are not persisted, unacknowledged consumers lose messages, and there is no retry semantics. A database-backed queue (a jobs table with status, next_run_at, and polling) is simple and avoids a Redis dependency, but polling adds latency, advisory locks are fiddly, and you end up reimplementing features BullMQ ships with — priorities, backoff, rate limiting, job events.

BullMQ uses Redis sorted sets for scheduling, Redis streams under the hood for reliable acknowledgement, and Lua scripts for atomic operations. The result is a queue that is:

  • Reliable: a job is never lost even if the worker crashes mid-execution (it re-appears in the active set after a stall timeout).
  • Observable: every state transition (waiting → active → completed / failed / delayed) is recorded.
  • Concurrent: multiple workers can share the same queue; Redis handles coordination.
  • Fast: Redis keeps everything in memory; throughput easily exceeds 10 000 jobs/second on modest hardware.

Installation and initial setup

npm install bullmq ioredis
npm install -D @types/node typescript tsx

BullMQ requires Redis 6.2 or later. For local development, the easiest path is Docker:

docker run -d \
  --name bullmq-redis \
  -p 6379:6379 \
  redis:7-alpine \
  redis-server --appendonly yes

The --appendonly yes flag enables Redis AOF persistence so the queue survives a Redis restart.

Create a shared Redis connection configuration. BullMQ recommends separate connections for queues and workers because some Redis operations block:

// src/lib/redis.ts
import { ConnectionOptions } from "bullmq";

export const redisConnection: ConnectionOptions = {
  host: process.env.REDIS_HOST ?? "localhost",
  port: parseInt(process.env.REDIS_PORT ?? "6379", 10),
  password: process.env.REDIS_PASSWORD,
  maxRetriesPerRequest: null, // required by BullMQ
};

The maxRetriesPerRequest: null setting is mandatory for BullMQ worker connections — it prevents ioredis from throwing on blocking commands that take longer than the default retry timeout.

Defining a typed job queue

BullMQ is written in TypeScript and exposes generics for job data and return values. Use these to keep your job payloads and results fully typed end-to-end.

// src/queues/email.ts
import { Queue } from "bullmq";
import { redisConnection } from "../lib/redis.js";

export interface SendEmailJobData {
  to: string;
  subject: string;
  templateId: string;
  variables: Record<string, string>;
  userId?: string;
}

export interface SendEmailJobResult {
  messageId: string;
  sentAt: string;
}

export const emailQueue = new Queue<SendEmailJobData, SendEmailJobResult>(
  "send-email",
  {
    connection: redisConnection,
    defaultJobOptions: {
      attempts: 3,
      backoff: {
        type: "exponential",
        delay: 2000, // first retry after 2 s, then 4 s, then 8 s
      },
      removeOnComplete: { count: 1000 }, // keep last 1000 completed jobs
      removeOnFail: { count: 5000 },     // keep last 5000 failed jobs
    },
  }
);

The defaultJobOptions set here apply to every job added to the queue unless overridden per-job. removeOnComplete and removeOnFail are critical for production: without them, Redis slowly fills up with job records.

Adding a job to the queue is a single method call:

// From any API route or service
import { emailQueue } from "./queues/email.js";

const job = await emailQueue.add("welcome-email", {
  to: "[email protected]",
  subject: "Welcome to the platform!",
  templateId: "welcome-v2",
  variables: {
    firstName: "Amara",
    activationLink: "https://app.example.com/activate?token=abc123",
  },
  userId: "user_01J3KX...",
});

console.log(`Job ${job.id} added to queue`);

The first argument to add is the job name — a label you can use to route different job types to different processors in the same worker.

Writing the worker

A Worker subscribes to a queue and processes jobs. Each job gets a dedicated invocation of the processor function.

// src/workers/email-worker.ts
import { Worker, Job } from "bullmq";
import { redisConnection } from "../lib/redis.js";
import type { SendEmailJobData, SendEmailJobResult } from "../queues/email.js";

async function processEmailJob(
  job: Job<SendEmailJobData, SendEmailJobResult>
): Promise<SendEmailJobResult> {
  const { to, subject, templateId, variables } = job.data;

  // Report progress so the UI can show a spinner
  await job.updateProgress(10);

  // Render the template (in a real system, call your template engine)
  const html = renderTemplate(templateId, variables);
  await job.updateProgress(40);

  // Send via your email provider (SendGrid, Resend, SES, etc.)
  const messageId = await sendEmail({ to, subject, html });
  await job.updateProgress(100);

  return {
    messageId,
    sentAt: new Date().toISOString(),
  };
}

function renderTemplate(
  templateId: string,
  variables: Record<string, string>
): string {
  // Placeholder — replace with your real template engine
  return `<p>Hello ${variables.firstName ?? "there"},</p><p>Template: ${templateId}</p>`;
}

async function sendEmail(params: {
  to: string;
  subject: string;
  html: string;
}): Promise<string> {
  // Placeholder — replace with your real email SDK call
  console.log(`Sending email to ${params.to}: ${params.subject}`);
  await new Promise((r) => setTimeout(r, 200)); // simulate network call
  return `msg_${Date.now()}`;
}

const worker = new Worker<SendEmailJobData, SendEmailJobResult>(
  "send-email",
  processEmailJob,
  {
    connection: redisConnection,
    concurrency: 5, // process up to 5 emails simultaneously
  }
);

worker.on("completed", (job, result) => {
  console.log(`Job ${job.id} completed. Message ID: ${result.messageId}`);
});

worker.on("failed", (job, error) => {
  console.error(`Job ${job?.id} failed: ${error.message}`);
});

worker.on("stalled", (jobId) => {
  console.warn(`Job ${jobId} stalled and will be requeued`);
});

The worker's concurrency option controls how many jobs run simultaneously within a single worker process. For I/O-bound jobs (network calls, database queries), a concurrency of 10–50 is typical. For CPU-bound jobs (image processing, PDF generation), set it to 1 or match the number of CPU cores and spawn multiple worker processes instead.

Priority queues

BullMQ supports job priorities out of the box. Lower numbers are higher priority (1 is the most urgent). This is implemented efficiently via Redis sorted sets.

// Urgent transactional email — processes before marketing emails
await emailQueue.add(
  "password-reset",
  { to: user.email, subject: "Reset your password", templateId: "password-reset", variables: { link } },
  { priority: 1 }
);

// Weekly newsletter — can wait
await emailQueue.add(
  "weekly-digest",
  { to: user.email, subject: "Your weekly digest", templateId: "digest", variables: {} },
  { priority: 10 }
);

Workers automatically pick higher-priority jobs first when multiple jobs are waiting. You do not need a separate queue per priority level.

Delayed jobs and scheduled (repeatable) jobs

Delay a job by specifying a number of milliseconds:

// Send a follow-up email 3 days after signup
await emailQueue.add(
  "onboarding-followup",
  { to: user.email, subject: "How are you getting on?", templateId: "followup", variables: {} },
  { delay: 3 * 24 * 60 * 60 * 1000 } // 72 hours in ms
);

Delayed jobs sit in the delayed state in Redis and are promoted to waiting automatically when their delay expires. BullMQ uses a Redis sorted set with timestamps as scores for this — no polling loop required.

For repeatable jobs (cron-style), BullMQ's scheduler uses cron expressions:

import { Queue } from "bullmq";

const reportQueue = new Queue("reports", { connection: redisConnection });

// Run every day at 08:00 UTC
await reportQueue.add(
  "daily-summary",
  { type: "daily", recipients: ["[email protected]"] },
  {
    repeat: {
      pattern: "0 8 * * *", // cron expression
      tz: "UTC",
    },
  }
);

// Run every 15 minutes
await reportQueue.add(
  "health-check-report",
  { type: "health" },
  { repeat: { every: 15 * 60 * 1000 } } // ms interval
);

Only one instance of each repeatable job is scheduled at a time, even if multiple producer processes call add with the same key. BullMQ deduplicates by the job name + repeat pattern.

To list or remove repeatable jobs:

const repeatableJobs = await reportQueue.getRepeatableJobs();
console.log(repeatableJobs);

// Remove a specific repeatable job
await reportQueue.removeRepeatableByKey(repeatableJobs[0].key);

Rate limiting with the rate limiter option

Some jobs must not exceed a throughput threshold — API providers impose rate limits, and you should not hammer a third-party service just because Redis can. BullMQ has a built-in rate limiter:

const smsQueue = new Queue("send-sms", { connection: redisConnection });

const smsWorker = new Worker(
  "send-sms",
  async (job) => {
    await sendSms(job.data.phone, job.data.message);
  },
  {
    connection: redisConnection,
    limiter: {
      max: 100,       // max 100 jobs
      duration: 1000, // per 1000 ms (i.e., 100/second)
    },
  }
);

Jobs that would exceed the rate are automatically delayed by the worker, not rejected. They return to the waiting state and will be picked up as soon as the window allows.

Flow producers: parent-child job trees

One of BullMQ's most powerful features is flow producers, which let you define trees of dependent jobs. A parent job completes only when all its children complete. This is ideal for pipelines: validate → transform → upload → notify.

import { FlowProducer } from "bullmq";

const flow = new FlowProducer({ connection: redisConnection });

const tree = await flow.add({
  name: "notify-user",
  queueName: "notifications",
  data: { userId: "user_01J3" },
  children: [
    {
      name: "generate-pdf",
      queueName: "pdf-generation",
      data: { reportId: "report_456" },
      children: [
        {
          name: "fetch-data",
          queueName: "data-fetching",
          data: { reportId: "report_456", source: "analytics-db" },
        },
      ],
    },
    {
      name: "send-email-notification",
      queueName: "send-email",
      data: {
        to: "[email protected]",
        subject: "Your report is ready",
        templateId: "report-ready",
        variables: {},
      },
    },
  ],
});

console.log("Flow job tree created:", tree.job.id);

The notify-user parent job will not start until both generate-pdf and send-email-notification are complete. The generate-pdf job will not start until fetch-data is complete. BullMQ manages this dependency graph in Redis automatically.

Workers for each queue process their jobs independently. A worker processing notifications checks whether all children have completed before running the parent processor.

Monitoring with BullMQ Board

Knowing what is happening inside your queues is essential for operations. The community-maintained @bull-board/api package provides a web dashboard that connects directly to your Redis instance and shows queue depths, job states, throughput graphs, and lets you retry or delete individual jobs.

npm install @bull-board/api @bull-board/express

// src/admin/dashboard.ts
import express from "express";
import { createBullBoard } from "@bull-board/api";
import { BullMQAdapter } from "@bull-board/api/bullMQAdapter.js";
import { ExpressAdapter } from "@bull-board/express";
import { emailQueue } from "../queues/email.js";
import { reportQueue } from "../queues/reports.js";

export function mountDashboard(app: express.Application) {
  const serverAdapter = new ExpressAdapter();
  serverAdapter.setBasePath("/admin/queues");

  createBullBoard({
    queues: [
      new BullMQAdapter(emailQueue),
      new BullMQAdapter(reportQueue),
    ],
    serverAdapter,
  });

  // Protect the dashboard in production
  app.use("/admin/queues", requireAdminAuth, serverAdapter.getRouter());
}

function requireAdminAuth(
  req: express.Request,
  res: express.Response,
  next: express.NextFunction
) {
  const token = req.headers["x-admin-token"];
  if (token !== process.env.ADMIN_TOKEN) {
    res.status(401).json({ error: "Unauthorized" });
    return;
  }
  next();
}

Mount the dashboard in your Express app:

import { mountDashboard } from "./admin/dashboard.js";
mountDashboard(app);

Navigate to http://localhost:3000/admin/queues to see the board. In production, protect this route with authentication — the dashboard can delete jobs and drain queues.

Handling failures gracefully

The default retry behaviour is often insufficient for production. Here is a more complete failure handling setup:

worker.on("failed", async (job, error) => {
  if (!job) return;

  const { attemptsMade, opts } = job;
  const maxAttempts = opts.attempts ?? 1;

  if (attemptsMade >= maxAttempts) {
    // Final failure — alert the team
    await alertSlack({
      message: `Job ${job.id} (${job.name}) permanently failed after ${attemptsMade} attempts`,
      error: error.message,
      jobData: job.data,
    });

    // Optionally, move critical job data to a dead-letter store for manual review
    await db.collection("failed_jobs").add({
      jobId: job.id,
      queueName: job.queueName,
      jobName: job.name,
      data: job.data,
      error: error.message,
      stack: error.stack,
      failedAt: new Date(),
    });
  }
});

For jobs that fail due to transient issues (network timeouts, rate limit responses), exponential backoff is the right default. For jobs that fail due to invariant violations (invalid data that will never succeed), you want to detect the condition early and move the job to a dead-letter queue rather than burning retry attempts:

async function processWithEarlyTermination(job: Job) {
  const { data } = job;

  // Validate data before doing expensive work
  if (!isValidEmailAddress(data.to)) {
    // Mark as failed immediately without retrying
    throw Object.assign(
      new Error(`Invalid email address: ${data.to}`),
      { failedOnPurpose: true }
    );
  }

  // ... rest of processing
}

You can inspect error.failedOnPurpose in the failed event handler and set the job's remaining attempts to 0 to prevent further retries.

Graceful shutdown

Worker processes must shut down gracefully — in-flight jobs should complete before the process exits, not be killed mid-execution.

async function shutdown() {
  console.log("Shutting down worker...");

  // Stop accepting new jobs
  await worker.close();

  // Give in-flight jobs time to complete (default: waits indefinitely)
  // Pass a timeout in ms to force-kill after that duration:
  // await worker.close(true); // force immediate close

  console.log("Worker shut down cleanly.");
  process.exit(0);
}

process.on("SIGTERM", shutdown);
process.on("SIGINT", shutdown);

Kubernetes sends SIGTERM before killing a pod. Listening for it and calling worker.close() ensures your worker drains its current jobs before the pod terminates. Set your pod's terminationGracePeriodSeconds to longer than your longest expected job duration.

Production architecture: multiple queues and worker pools

For a real application, separate concerns across multiple queues and deploy independent worker pods for each:

Producer (API server)
  ↓ adds jobs to Redis
  ├── Queue: send-email
  ├── Queue: generate-pdf
  ├── Queue: process-upload
  └── Queue: send-notification

Workers (separate processes/pods)
  ├── email-worker (concurrency: 20, 2 replicas)
  ├── pdf-worker (concurrency: 2, 4 replicas, more CPU)
  ├── upload-worker (concurrency: 5, 3 replicas)
  └── notification-worker (concurrency: 50, 1 replica with rate limiter)

This isolation means:

  • A spike in PDF jobs does not delay email delivery.
  • You can scale PDF workers independently when demand increases.
  • A crash in the upload worker does not affect any other queue.

The queue names are just Redis keys, so all workers share the same Redis instance and the same connection configuration.

Metrics and observability

BullMQ emits events for every job state transition. Hook into these to publish metrics to your observability stack:

import { QueueEvents } from "bullmq";

const emailQueueEvents = new QueueEvents("send-email", {
  connection: redisConnection,
});

emailQueueEvents.on("completed", ({ jobId, returnvalue }) => {
  metrics.increment("jobs.completed", { queue: "send-email" });
  metrics.histogram("job.duration", /* calculate from timestamps */ 0, {
    queue: "send-email",
  });
});

emailQueueEvents.on("failed", ({ jobId, failedReason }) => {
  metrics.increment("jobs.failed", { queue: "send-email" });
});

emailQueueEvents.on("stalled", ({ jobId }) => {
  metrics.increment("jobs.stalled", { queue: "send-email" });
});

QueueEvents uses Redis pub/sub to receive events from any worker connected to the same queue, so you can run this in a dedicated metrics process without touching the worker code.

Common pitfalls

Not setting removeOnComplete and removeOnFail. By default, BullMQ keeps every completed and failed job record in Redis forever. A busy queue will exhaust Redis memory over time. Always set these options.

Sharing ioredis connections between queues and workers. BullMQ requires workers to use blocking Redis commands. Use separate IORedis instances (or let BullMQ create its own) for the Queue and Worker classes.

Setting concurrency too high for CPU-bound work. If your processor function does CPU-intensive work (image manipulation, cryptography), high concurrency means all jobs fight over the same event loop and CPU. Spawn multiple worker processes with concurrency 1 instead.

Not handling the stalled event. A stalled job is one where the worker stopped sending keepalive signals (because the process crashed or hung). BullMQ requeues stalled jobs automatically, but if this happens repeatedly, it indicates a worker that is timing out. Monitor stall counts to detect issues early.

Ignoring return values from async add calls. If you fire-and-forget emailQueue.add(...) without awaiting it, you lose the job ID and cannot tell whether the job was actually enqueued. Always await the call and log or store the job ID.

BullMQ with Redis gives you a battle-tested foundation for background job processing that scales from a single-developer side project to a system handling millions of jobs per day. The combination of TypeScript generics for type safety, Redis persistence for durability, and a rich feature set for concurrency control and scheduling makes it the right choice for the vast majority of Node.js backend applications.

Comments

No comments yet. Be the first!

Sign in to leave a comment.