How to Set Up BullMQ Workers: Patterns, Concurrency, and Error Handling

BullMQ is the most widely adopted Redis-backed job queue for Node.js, with over 5.9 million weekly downloads and 8,900+ GitHub stars. At the heart of any BullMQ-based system is the Worker — the component that pulls jobs from a queue, processes them, handles failures, and reports progress. Getting the Worker setup right determines whether your background job pipeline is reliable, scalable, and production-ready.

In this post, we'll cover everything you need to set up BullMQ Workers in Node.js and TypeScript: basic instantiation, concurrency patterns, Redis connection management, error handling with retries and backoff, sandboxed processors for CPU-heavy work, graceful shutdown, and monitoring with QueueHub for real-time visibility.

Prerequisites

Before diving in, make sure you have:

• Node.js 18+ and npm, yarn, pnpm, or bun
• A running Redis instance — locally (Redis or Valkey), via Docker, or a cloud provider (Redis Cloud, AWS ElastiCache, Upstash)
• BullMQ installed: npm install bullmq or bun add bullmq
• IORedis — BullMQ uses IORedis internally and includes it as a peer dependency
• Basic familiarity with TypeScript (optional but highly recommended)

If you don't have Redis locally, spin one up quickly:

docker run -d -p 6379:6379 redis:7-alpine

BullMQ Worker Basics

What Is a Worker?

A Worker in BullMQ is the processing unit that pulls jobs from a queue and executes them via a processor function. Think of it as the consumer half of a producer-consumer pattern — the Queue adds jobs, the Worker processes them.

Under the hood, each Worker uses a blocking BRPOPLPUSH-style mechanism on Redis. This means workers do not poll — they block efficiently on Redis until a job is available, then immediately claim and process it. Each Worker instance consumes at least one Redis connection (plus a shared blocking connection duplicated internally).

Creating Your First Worker

Here's the simplest possible Worker in TypeScript:

import { Worker, Job } from "bullmq";

const worker = new Worker(
  "email",
  async (job: Job) => {
    console.log(`Processing job ${job.id}: ${job.name}`);
    const { to, subject, body } = job.data;
    await sendEmail(to, subject, body);
    return { sent: true, to };
  },
  {
    connection: {
      host: "localhost",
      port: 6379,
    },
  }
);

worker.on("completed", (job: Job, returnvalue: any) => {
  console.log(`Job ${job.id} completed`, returnvalue);
});

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

// CRITICAL: Always attach an 'error' listener to prevent worker stoppage
worker.on("error", (err: Error) => {
  console.error("Worker error:", err);
});

Three things to notice here:

1. The processor function is the second argument — it receives the Job instance and returns a value or throws an error.
2. We attach event listeners for completed, failed, and error.
3. The error event listener is critical. Without it, BullMQ will crash the worker on Redis connection errors. Always add one.

The Processor Function

The processor function is the core of your worker. It must be an async function (or return a Promise). It receives the Job instance, an optional token string, and an optional AbortSignal for cancellation:

const worker = new Worker<string, boolean>(
  "my-queue",
  async (job: Job<string>, token?: string, signal?: AbortSignal) => {
    // Check for cancellation
    if (signal?.aborted) {
      throw new Error("Job was cancelled");
    }

    await job.updateProgress(25);
    // ... processing ...
    await job.updateProgress(50);
    // ... more processing ...
    await job.updateProgress(100);

    return true;
  },
  { connection }
);

The job's return value is stored as job.returnvalue and emitted in the completed event. Use job.updateProgress() to report progress to monitoring dashboards.

TypeScript Generics with Worker

BullMQ Workers are fully typed with generics. You can type the job data payload and the return value:

interface EmailData {
  to: string;
  subject: string;
  body: string;
}

interface EmailResult {
  sent: boolean;
  messageId: string;
}

const worker = new Worker<EmailData, EmailResult>(
  "email",
  async (job: Job<EmailData, EmailResult>) => {
    const result = await sendEmail(job.data);
    return result;
  },
  { connection }
);

Redis Connections

Connection Options

BullMQ uses IORedis under the hood. Connection options are passed directly to the IORedis constructor. Here are common deployment patterns:

// Local Redis
const worker = new Worker("queue", processor, {
  connection: { host: "localhost", port: 6379 },
});

// Redis with TLS (common with cloud providers)
const worker = new Worker("queue", processor, {
  connection: {
    host: "myredis.tls.com",
    port: 6380,
    tls: {},
    password: "my-password",
  },
});

// AWS ElastiCache
const worker = new Worker("queue", processor, {
  connection: {
    host: "my-cluster.xxxxxx.region.cache.amazonaws.com",
    port: 6379,
  },
});

// Valkey (compatible Redis fork)
const worker = new Worker("queue", processor, {
  connection: {
    host: "my-valkey.instance",
    port: 6379,
  },
});

Reusing Redis Connections (Best Practice)

Never create a separate Redis connection for every Queue, Worker, and QueueEvents instance. Share one IORedis instance across all BullMQ components:

import IORedis from "ioredis";
import { Queue, Worker, QueueEvents } from "bullmq";

const connection = new IORedis({
  host: "localhost",
  port: 6379,
  maxRetriesPerRequest: null,
});

const queue = new Queue("my-queue", { connection });
const worker = new Worker("my-queue", async (job) => {
  // process job
}, { connection });
const queueEvents = new QueueEvents("my-queue", { connection });

Setting maxRetriesPerRequest: null is important — it prevents IORedis from automatically retrying commands that fail (BullMQ handles retries internally).

Concurrency

The Concurrency Option

By default, a BullMQ Worker processes one job at a time. The concurrency option controls how many jobs the worker processes simultaneously:

const worker = new Worker("queue", processor, {
  connection,
  concurrency: 10,
});

This worker will process up to 10 jobs concurrently. The right concurrency setting depends on your workload:

• I/O-bound tasks (API calls, database queries, file uploads): High concurrency (20-50+) works well because the event loop isn't blocked during I/O waits.
• CPU-bound tasks (image processing, video transcoding, data transformation): Keep concurrency low (1-4) or use sandboxed processors (covered below).
• Memory-intensive tasks: Low concurrency (1-2) to avoid OOM errors.

Limiter (Rate Limiting)

For integration with rate-limited APIs, combine concurrency with a limiter:

const worker = new Worker("api-queue", processor, {
  connection,
  concurrency: 10,
  limiter: {
    max: 5,   // max 5 jobs
    duration: 1000,  // per second
  },
});

This ensures no more than 5 jobs are processed per second across all concurrent slots.

Error Handling

Retry Strategies with Backoff

BullMQ's retry mechanism is one of its most powerful features. You configure retries per job when adding to the queue:

// Per-job retry configuration
await queue.add("email", data, {
  attempts: 3,
  backoff: {
    type: "exponential",
    delay: 2000,
  },
});

// Or set as defaults on the Queue
const queue = new Queue("email", {
  connection,
  defaultJobOptions: {
    attempts: 3,
    backoff: { type: "exponential", delay: 1000 },
  },
});

Backoff Strategies

BullMQ provides three backoff strategies:

Strategy	Behavior	Use Case
fixed	Same delay between each retry	Transient issues, rate limits
exponential	2^(attempts-1) × delay	Server errors, network blips
custom	User-defined function	Complex retry logic

Here's how to use them:

// Fixed with jitter
await queue.add("job", data, {
  attempts: 5,
  backoff: { type: "fixed", delay: 5000, jitter: 0.3 },
});

// Custom backoff strategy (defined on the Worker)
const worker = new Worker("queue", processor, {
  settings: {
    backoffStrategy: (attemptsMade: number, type?: string, err?: Error, job?: Job) => {
      // Linear: 1s, 2s, 3s, ...
      return attemptsMade * 1000;
    },
  },
});

// Job opts reference it as 'custom'
await queue.add("job", data, {
  attempts: 3,
  backoff: { type: "custom" },
});

Stalled Jobs

A job becomes stalled when a worker fails to renew its processing lock. This typically happens when:

• The worker's event loop is blocked by CPU-heavy processing
• The worker crashed unexpectedly
• Network issues prevent the lock renewal heartbeat

BullMQ detects stalled jobs and moves them back to the waiting queue for re-processing. After maxStalledCount (default: 1), the job moves to the failed state permanently.

Prevention: Use sandboxed processors for CPU-heavy work, keep concurrency at reasonable levels, and avoid blocking the event loop with synchronous operations.

The Failed Event

Listen to the failed event for detailed error tracking and alerting:

worker.on("failed", (job: Job | undefined, error: Error, prev: string) => {
  console.error(`Job ${job?.id} failed after ${job?.attemptsMade} attempts`);
  console.error(`Error: ${error.message}`);
  console.error(`Previous state: ${prev}`);

  // Send to monitoring/alerting
  await notifyFailure(job, error);
});

Stopping Retries Conditionally

For non-retryable errors — validation failures, authentication errors, bad input — you can stop retrying by returning -1 from a custom backoff strategy:

import { NonRetryableError } from "bullmq";

const worker = new Worker("queue", processor, {
  settings: {
    backoffStrategy: (attemptsMade, type, err) => {
      if (err instanceof NonRetryableError) {
        return -1; // Stop retrying, move to failed permanently
      }
      return attemptsMade * 5000;
    },
  },
});

This is a clean pattern for distinguishing transient failures (retry) from permanent ones (move to failed immediately).

Sandboxed Workers (Separate Process)

When to Use Sandboxed Processors

For CPU-heavy or memory-intensive work, running the processor in the same Node.js process as the worker can block the event loop, causing stalled jobs and degraded performance. Sandboxed processors solve this by running the job in a separate process (or thread).

Use sandboxed processors when:

• Performing CPU-heavy operations like image processing, PDF generation, or video transcoding
• Running memory-intensive computations
• Isolating unstable third-party code that might crash
• Preventing stalled jobs caused by lock renewal delays during heavy processing

Creating a Sandboxed Processor

Create a separate file for the processor:

// processor.ts (separate file)
import { SandboxedJob } from "bullmq";

export default async (job: SandboxedJob) => {
  // CPU-heavy work here — won't affect the main worker process
  const result = await performHeavyComputation(job.data);
  return result;
};

Then reference it in your Worker by file path:

// worker.ts
import path from "path";
import { Worker } from "bullmq";

const processorFile = path.join(__dirname, "processor.js");

const worker = new Worker("heavy-queue", processorFile, {
  connection,
  concurrency: 2, // Keep low for CPU-bound work
  useWorkerThreads: true, // Use worker_threads instead of child_process
});

Worker Threads vs Child Processes

By default, BullMQ spawns a new child process per sandboxed job. Setting useWorkerThreads: true switches to Node.js worker_threads, which are lighter (shared memory space, no process overhead). For ESM compatibility on Windows, use the URL-based path:

const processorUrl = new URL("./processor.js", import.meta.url);
const worker = new Worker("heavy-queue", processorUrl.href, {
  connection,
  useWorkerThreads: true,
});

Worker Lifecycle & Events

Worker Events Reference

BullMQ Workers emit several lifecycle events you can listen to:

worker.on("completed", (job: Job, returnvalue: any) => {
  // Job finished successfully
});

worker.on("failed", (job: Job | undefined, error: Error, prev: string) => {
  // Job processing threw an error
});

worker.on("progress", (job: Job, progress: number | object) => {
  // Job reported progress via job.updateProgress()
});

worker.on("drained", () => {
  // Queue is empty — no more jobs to process
});

worker.on("error", (err: Error) => {
  // Connection or internal error — ALWAYS listen to this
});

worker.on("active", (job: Job) => {
  // Job has started processing
});

worker.on("closing", (msg: string) => {
  // Worker is shutting down
});

Global Events with QueueEvents

For monitoring across all worker instances (useful in multi-process or distributed deployments), use QueueEvents:

import { QueueEvents } from "bullmq";

const queueEvents = new QueueEvents("my-queue", { connection });

queueEvents.on("completed", ({ jobId, returnvalue }) => {
  console.log(`Job ${jobId} completed globally`);
});

queueEvents.on("failed", ({ jobId, failedReason }) => {
  console.log(`Job ${jobId} failed: ${failedReason}`);
});

queueEvents.on("progress", ({ jobId, data }) => {
  console.log(`Job ${jobId} progress:`, data);
});

QueueEvents uses Redis streams, which are reliable, auto-trimmed, and survive disconnections. This makes it ideal for centralized monitoring.

Graceful Shutdown

Why Graceful Shutdown Matters

When your application receives a shutdown signal (SIGTERM from Kubernetes, SIGINT from Ctrl+C), you need to:

1. Stop accepting new jobs
2. Let active jobs complete (or move them back to the waiting queue)
3. Cleanly close Redis connections

Without graceful shutdown, you risk abandoning mid-processing jobs and leaving dangling Redis connections.

Shutdown Sequence

async function shutdown(worker: Worker, queue: Queue, signal: string) {
  console.log(`Received ${signal}. Starting graceful shutdown...`);

  // 1. Stop accepting new jobs — waits for active jobs by default
  await worker.close();

  // 2. Close the queue connection
  await queue.close();

  console.log("Graceful shutdown complete");
  process.exit(0);
}

// In your application
const worker = new Worker("queue", processor, { connection });
const queue = new Queue("queue", { connection });

process.on("SIGTERM", () => shutdown(worker, queue, "SIGTERM"));
process.on("SIGINT", () => shutdown(worker, queue, "SIGINT"));

Close Options

// Force close — don't wait for active jobs
await worker.close(true); // force = true

// Close with custom timeout (default: wait indefinitely)
await worker.close(false, 30_000); // 30s timeout for active jobs

Monitoring Workers with QueueHub

The Visibility Gap

Even with perfect code, production workers can encounter issues that are invisible without proper monitoring: stuck jobs, stalled workers, memory leaks, concurrency bottlenecks, and failure spikes. Logs alone don't give you a real-time picture of queue health.

What QueueHub Shows

QueueHub provides a dedicated dashboard for your BullMQ queues that reveals:

• Live Worker View: See active workers, their concurrency settings, and which jobs they're processing in real time
• Job Detail View: Inspect job data, return values, error stack traces, and retry history
• Queue Overview: Throughput charts, failure rates, and processing latency at a glance
• Multi-Backend Support: Connect to any Redis deployment (local, TLS, Valkey, AWS ElastiCache) via connection settings or agent tunneling for private Redis instances
• Team Collaboration: Share queue visibility across your whole team with multi-org support and team invitations

Connecting QueueHub to Your Workers

Point QueueHub at your Redis instance — it works across local development, staging, and production environments. Supports TLS-secured Redis, AWS ElastiCache, Valkey, and private backends via agent tunneling.

Production Best Practices — Summary

1. Always attach an error listener — prevents silent worker stoppage when Redis connections hiccup
2. Reuse Redis connections — share one IORedis instance across Queue, Worker, and QueueEvents
3. Set maxRetriesPerRequest: null on reused IORedis connections to avoid automatic reconnect issues
4. Match concurrency to workload — high for I/O-bound tasks, low for CPU-bound or memory-intensive work
5. Use exponential backoff for transient failures, custom backoff with -1 return for non-retryable errors
6. Sandbox CPU-heavy processors — prevents stalled jobs from lock renewal delays
7. Implement graceful shutdown — handle SIGTERM and SIGINT to avoid job abandonment
8. Set defaultJobOptions on the Queue to avoid repeating retry/backoff config per job
9. Monitor with QueueHub — real-time visibility into worker health, job failures, and queue performance
10. Use QueueEvents for global monitoring across all worker instances in distributed deployments

Conclusion

BullMQ Workers are the backbone of reliable background job processing. Proper setup involves more than just a processor function — you need to manage Redis connections efficiently, tune concurrency to your workload, implement robust error handling with retries and backoff, use sandboxed processors for CPU-heavy work, and handle graceful shutdown signals.

The patterns covered in this post will serve you well in production. For your next steps:

• Try these patterns in your own project
• Explore the official BullMQ documentation for advanced topics like rate limiting, migrations, and job schedulers
• Set up QueueHub for real-time visibility into your queues — don't debug worker issues blind

Appendix: Quick Reference

Worker Constructor Signature

new Worker<TData = any, TReturn = any, TName extends string = string>(
  name: string,
  processor:
    | string
    | ((
        job: Job<TData, TReturn, TName>,
        token?: string,
        signal?: AbortSignal
      ) => Promise<TReturn>),
  opts?: WorkerOptions
);

interface WorkerOptions {
  connection?: IORedis.Redis | IRedisClient | ConnectionOptions;
  concurrency?: number;
  autorun?: boolean;
  limiter?: RateLimiterOptions;
  skipStalledCheck?: boolean;
  stalledInterval?: number;
  maxStalledCount?: number;
  removeOnComplete?: KeepJobs;
  removeOnFail?: KeepJobs;
  useWorkerThreads?: boolean;
  settings?: WorkerSettings;
}

Common Patterns Quick Copy

// Minimal setup
new Worker("queue", async (job) => { /* ... */ }, { connection });

// With concurrency + rate limit
new Worker("queue", async (job) => { /* ... */ }, {
  connection,
  concurrency: 10,
  limiter: { max: 5, duration: 1000 },
});

// With retry-friendly jobs
queue.add("job", data, {
  attempts: 5,
  backoff: { type: "exponential", delay: 1000 },
});

// Graceful shutdown
await worker.close();
await queue.close();

---

Last updated: June 2026 — compatible with BullMQ v5+