Message Queues with BullMQ in Node.js

Your monolith is being split into microservices, and the first challenge is communication: Service A needs to notify Service B when a user signs up, but they should never depend on each other being available simultaneously. A message queue decouples them — Service A publishes an event, Service B processes it when ready. BullMQ has become the standard message queue library for Node.js microservices, combining Redis reliability with job-specific features like retries, delays, and rate limiting. This article covers the producer-consumer pattern, job flows, and monitoring production queues.

Why BullMQ for Production Message Queues

BullMQ is a Redis-backed job queue library for Node.js that handles millions of jobs daily in production. Unlike simple in-memory queues, BullMQ provides persistence, delayed jobs, rate limiting, and concurrency control. It's battle-tested at scale, used by companies like Slack and Discord. The key differentiator is its use of Redis streams for reliability—jobs survive crashes and restarts. For any system requiring background processing (email, image processing, webhooks), BullMQ is the go-to choice. Avoid alternatives like Kue (unmaintained) or Bee-Queue (limited features). BullMQ's maturity and active development make it production-ready.

npm install bullmq ioredis

Setting Up a Queue and Worker

A BullMQ queue holds jobs, and workers process them. The queue is a Redis-backed data structure; workers are Node.js processes that poll the queue. Here's a minimal setup: create a queue with a name, then a worker that processes jobs. The worker's concurrency option controls how many jobs run in parallel. Always handle errors in the worker—uncaught exceptions crash the process. Use the failed event for monitoring. The queue can also be used to add jobs with options like delay, priority, and attempts.

const { Queue, Worker } = require('bullmq');
const Redis = require('ioredis');

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

const emailQueue = new Queue('email', { connection });

const worker = new Worker('email', async job => {
  console.log(`Processing job ${job.id}: ${job.data.to}`);
  // Simulate email send
  await sendEmail(job.data);
}, { connection, concurrency: 5 });

worker.on('completed', job => console.log(`Job ${job.id} completed`));
worker.on('failed', (job, err) => console.error(`Job ${job.id} failed: ${err.message}`));

async function sendEmail(data) {
  // actual email sending logic
  return true;
}

module.exports = { emailQueue, worker };

Adding Jobs with Options

Jobs can be added with various options to control execution: delay for scheduled tasks, attempts for retry logic, backoff for exponential backoff, priority for ordering, and removeOnComplete to auto-clean. Use jobId for idempotency—if a job with the same ID exists, it won't be duplicated. This is critical for preventing duplicate payments or emails. Always set a reasonable attempts count (e.g., 3) with exponential backoff to handle transient failures.

const { emailQueue } = require('./queue-setup');

async function addEmailJob(to, subject, body) {
  await emailQueue.add('send-email', { to, subject, body }, {
    attempts: 3,
    backoff: { type: 'exponential', delay: 2000 },
    removeOnComplete: { age: 3600, count: 1000 },
    jobId: `email:${to}:${Date.now()}`,
  });
}

addEmailJob('user@example.com', 'Welcome!', 'Hello...');

Handling Job Failures and Retries

BullMQ automatically retries failed jobs based on the attempts option. The worker's failed event fires after all retries are exhausted. For transient errors (e.g., network timeout), retry with backoff. For permanent errors (e.g., invalid data), mark the job as failed immediately by throwing a specific error. Use the removeOnFail option to auto-clean failed jobs. Monitor failed jobs via a dashboard or alerts. Never silently swallow errors—always log them.

const { Worker } = require('bullmq');
const Redis = require('ioredis');

const connection = new Redis();

const worker = new Worker('email', async job => {
  if (!job.data.to.includes('@')) {
    throw new Error('Invalid email'); // permanent failure, no retry
  }
  // Simulate transient failure
  if (Math.random() < 0.3) throw new Error('Network timeout');
  console.log(`Sent to ${job.data.to}`);
}, { connection, concurrency: 10 });

worker.on('failed', (job, err) => {
  console.error(`Job ${job.id} failed after ${job.attemptsMade} attempts: ${err.message}`);
});

Rate Limiting and Concurrency Control

BullMQ supports per-queue rate limiting to avoid overwhelming downstream services. Use the limiter option on the queue: max (max jobs per interval) and duration (interval in ms). For example, limit to 10 jobs per second. Concurrency on the worker controls how many jobs run in parallel. Combine both: set worker concurrency high, and let the limiter throttle. This is crucial for APIs with rate limits (e.g., SendGrid, Twilio).

const { Queue, Worker } = require('bullmq');
const Redis = require('ioredis');

const connection = new Redis();

const emailQueue = new Queue('email', {
  connection,
  limiter: { max: 10, duration: 1000 } // 10 jobs per second
});

const worker = new Worker('email', async job => {
  // send email
}, { connection, concurrency: 20 });

// Add jobs as usual

Scheduled and Delayed Jobs

BullMQ supports delayed jobs via the delay option (in ms). For recurring schedules (e.g., daily report), use the QueueScheduler or a separate cron worker. BullMQ doesn't have built-in cron; use node-cron or bullmq's repeatable jobs. Repeatable jobs are added with a repeat option specifying a cron pattern. They are stored in Redis and re-added automatically. Be careful with timezone—cron patterns are UTC by default.

const { Queue } = require('bullmq');
const Redis = require('ioredis');

const connection = new Redis();
const reportQueue = new Queue('reports', { connection });

// Add a repeatable job every day at midnight UTC
async function scheduleDailyReport() {
  await reportQueue.add('daily-report', { type: 'sales' }, {
    repeat: { pattern: '0 0 * * *' },
    jobId: 'daily-sales-report'
  });
}

scheduleDailyReport();

Monitoring and Observability

BullMQ provides events (completed, failed, progress, waiting) for real-time monitoring. Use Queue#getJobs to inspect queue state. For production, integrate with dashboards like Bull Board (a UI for Bull/BullMQ). Log job lifecycle events with structured logging (e.g., Pino). Track metrics: queue depth, processing time, failure rate. Use Redis INFO to monitor memory. Set up health checks: if queue depth exceeds threshold, alert.

const { Queue } = require('bullmq');
const Redis = require('ioredis');

const connection = new Redis();
const emailQueue = new Queue('email', { connection });

async function monitor() {
  const [waiting, active, completed, failed] = await Promise.all([
    emailQueue.getWaitingCount(),
    emailQueue.getActiveCount(),
    emailQueue.getCompletedCount(),
    emailQueue.getFailedCount()
  ]);
  console.log({ waiting, active, completed, failed });
}

setInterval(monitor, 5000);

Graceful Shutdown and Job Persistence

When shutting down a worker, you must wait for active jobs to complete to avoid losing work. BullMQ's worker.close() returns a promise that resolves after all active jobs finish. Use process signals (SIGTERM, SIGINT) to trigger graceful shutdown. For queues, persistence is automatic via Redis. However, if Redis goes down, jobs may be lost if not persisted to disk. Configure Redis appendonly yes for durability. In Kubernetes, use preStop hooks to drain workers.

const { Worker } = require('bullmq');
const Redis = require('ioredis');

const connection = new Redis();
const worker = new Worker('email', async job => {
  // process
}, { connection });

async function shutdown() {
  console.log('Shutting down worker...');
  await worker.close();
  await connection.quit();
  process.exit(0);
}

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

Scaling Workers Horizontally

BullMQ workers are stateless and can be scaled horizontally by running multiple instances. Each worker polls the same Redis queue. Use the concurrency option per worker to control parallelism. For high throughput, run many worker processes (e.g., one per CPU core). Use a process manager like PM2 or Kubernetes to manage worker instances. Be aware of Redis connection limits—each worker uses one connection. Use connection pooling with ioredis Cluster for large deployments.

const cluster = require('cluster');
const os = require('os');

if (cluster.isMaster) {
  const numWorkers = os.cpus().length;
  console.log(`Forking ${numWorkers} workers`);
  for (let i = 0; i < numWorkers; i++) {
    cluster.fork();
  }
  cluster.on('exit', (worker) => {
    console.log(`Worker ${worker.process.pid} died, restarting`);
    cluster.fork();
  });
} else {
  require('./worker'); // worker code
}

Advanced: Job Dependencies and Flows

BullMQ supports job dependencies via the parent option or using flows. A flow is a set of jobs with dependencies—a parent job waits for children to complete. This is useful for complex workflows like order processing (payment -> inventory -> shipping). Use QueueFlow to create flows. Children jobs can be added with children option. The parent job completes only after all children succeed. If any child fails, the parent fails. This pattern replaces ad-hoc orchestration.

const { QueueFlow } = require('bullmq');
const Redis = require('ioredis');

const connection = new Redis();
const flow = new QueueFlow(connection);

async function createOrderFlow(orderId) {
  const parent = await flow.add('order', { orderId }, {
    children: [
      { name: 'payment', data: { orderId }, queueName: 'payment' },
      { name: 'inventory', data: { orderId }, queueName: 'inventory' }
    ]
  });
  console.log(`Parent job ${parent.id} created`);
}

createOrderFlow('123');

Production Pitfalls and Best Practices

Common pitfalls: forgetting to set removeOnComplete leading to Redis memory bloat; not handling worker crashes (use stalledInterval to detect stalled jobs); using default Redis config without persistence; not monitoring queue depth. Best practices: always use environment variables for Redis connection; set stalledInterval to a reasonable value (default 30s); use jobId for idempotency; log job payloads (but sanitize PII); test failure scenarios with chaos engineering.

const { Queue, Worker } = require('bullmq');
const Redis = require('ioredis');

const connection = new Redis({
  host: process.env.REDIS_HOST || 'localhost',
  port: process.env.REDIS_PORT || 6379,
  password: process.env.REDIS_PASSWORD,
  maxRetriesPerRequest: null,
  enableReadyCheck: false
});

const queue = new Queue('myqueue', {
  connection,
  defaultJobOptions: {
    attempts: 3,
    backoff: { type: 'exponential', delay: 1000 },
    removeOnComplete: { age: 86400, count: 10000 },
    removeOnFail: { age: 604800 }
  }
});

const worker = new Worker('myqueue', async job => {
  // process
}, { connection, stalledInterval: 30000 });

Conclusion: BullMQ in the Real World

BullMQ is a robust, feature-rich queue system that scales from a single process to a cluster of workers. Its Redis-backed design provides persistence, reliability, and performance. By following the patterns in this article—proper job options, error handling, monitoring, and graceful shutdown—you can build production-grade background job systems. Remember: queues are not just for async tasks; they decouple components and improve resilience. Start simple, add features as needed, and always monitor.

// Full production setup
const { Queue, Worker, QueueScheduler } = require('bullmq');
const Redis = require('ioredis');

const connection = new Redis(process.env.REDIS_URL);

const queue = new Queue('tasks', { connection });
const scheduler = new QueueScheduler('tasks', { connection });

const worker = new Worker('tasks', async job => {
  // process
}, { connection, concurrency: 10 });

worker.on('completed', job => console.log(`Job ${job.id} done`));
worker.on('failed', (job, err) => console.error(`Job ${job.id} failed: ${err.message}`));

async function shutdown() {
  await worker.close();
  await scheduler.close();
  await queue.close();
  await connection.quit();
}

process.on('SIGTERM', shutdown);

Bull Board UI Dashboard

Bull Board is a UI dashboard for monitoring and managing Bull and BullMQ queues. It provides real-time visibility into job statuses, retries, and queue metrics. To integrate, install @bull-board/api and @bull-board/express. Mount the router on your Express app. You can customize authentication, queue display, and locale. For production, secure the dashboard behind an auth proxy or use the built-in middleware. Bull Board supports multiple queues, job data inspection, and manual job retries or removal. It's invaluable for debugging and operations.

const { createBullBoard } = require('@bull-board/api');
const { BullMQAdapter } = require('@bull-board/api/bullMQAdapter');
const { ExpressAdapter } = require('@bull-board/express');
const { Queue } = require('bullmq');

const serverAdapter = new ExpressAdapter();
serverAdapter.setBasePath('/admin/queues');

const queue = new Queue('myQueue', { connection });

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

app.use('/admin/queues', serverAdapter.getRouter());

Queue Pausing and Draining

BullMQ allows pausing and draining queues for maintenance or graceful shutdown. Pausing a queue prevents new jobs from being processed but keeps them in the queue. Draining removes all jobs (waiting, delayed, active) from the queue. Use queue.pause() and queue.drain() with optional delayed flag. Draining is destructive; use with caution. Pausing is useful for deployments: pause workers, let active jobs finish, then deploy new code. Resume with queue.resume(). For draining, consider moving jobs to a dead letter queue first.

const { Queue } = require('bullmq');
const queue = new Queue('myQueue', { connection });

// Pause queue (prevents new jobs from being processed)
await queue.pause();

// Drain all jobs (waiting, delayed, active)
await queue.drain();

// Drain only delayed jobs
await queue.drain({ delayed: true });

// Resume queue
await queue.resume();

OpenTelemetry Integration with BullMQ

OpenTelemetry provides observability through traces and metrics. BullMQ supports OpenTelemetry via the @opentelemetry/instrumentation-bullmq package. It automatically creates spans for job processing, queue operations, and worker lifecycle. Integrate with your existing OpenTelemetry setup (e.g., Jaeger, Zipkin, or cloud providers). This enables distributed tracing across microservices, showing how jobs flow through queues. To use, install the instrumentation and register it before creating queues or workers. Configure sampling rates to control overhead. Traces include job ID, queue name, and attempt number.

const { NodeTracerProvider } = require('@opentelemetry/sdk-trace-node');
const { registerInstrumentations } = require('@opentelemetry/instrumentation');
const { BullMQInstrumentation } = require('@opentelemetry/instrumentation-bullmq');

const provider = new NodeTracerProvider();
provider.register();

registerInstrumentations({
  instrumentations: [
    new BullMQInstrumentation({
      // Optional: hook to enrich spans
      requestHook: (span, job) => {
        span.setAttribute('job.id', job.id);
      },
    }),
  ],
});

// Now create queues and workers as usual

Dead Letter Queue Patterns

A dead letter queue (DLQ) stores jobs that have failed after exhausting retries. BullMQ doesn't have a built-in DLQ, but you can implement one easily. After max retries, move the failed job to a separate queue. Use the worker's failed event or a custom handler. Store the original job data and error details. Optionally, set up a separate worker to process the DLQ for manual inspection or reprocessing. This prevents poison messages from blocking the main queue. For critical jobs, alert when a job enters the DLQ.

const { Queue, Worker } = require('bullmq');

const mainQueue = new Queue('main', { connection });
const dlq = new Queue('dlq', { connection });

const worker = new Worker('main', async job => {
  // process job
}, {
  connection,
  maxStalledCount: 3,
});

worker.on('failed', async (job, err) => {
  if (job.attemptsMade >= job.opts.attempts) {
    await dlq.add(job.name, job.data, {
      jobId: job.id,
      attempts: 1, // retry DLQ processing once
    });
  }
});

// DLQ worker for manual reprocessing
const dlqWorker = new Worker('dlq', async job => {
  // Inspect and decide to reprocess or discard
  console.error('DLQ job:', job.id, job.data);
  // Optionally re-add to main queue
  await mainQueue.add(job.name, job.data);
}, { connection });

Unit Testing Workers

Unit testing BullMQ workers requires isolating job processing logic from Redis. Use bullmq's Job class to create mock jobs. Instantiate a worker with a mock connection or use ioredis-mock. For pure logic, extract the processor function and test it directly. Test job options, error handling, and retries. Use jest or mocha. For integration tests, use a real Redis instance (e.g., via Docker) but keep tests isolated. Mock the Worker constructor to avoid connecting to Redis in unit tests.

const { Job } = require('bullmq');
const { processJob } = require('./worker');

jest.mock('bullmq', () => ({
  ...jest.requireActual('bullmq'),
  Worker: jest.fn().mockImplementation(() => ({
    on: jest.fn(),
    close: jest.fn(),
  })),
}));

describe('processJob', () => {
  it('should process job data correctly', async () => {
    const data = { userId: 123 };
    const job = new Job({
      queue: { name: 'test' },
      data,
      opts: { attempts: 3 },
    });
    const result = await processJob(job);
    expect(result).toEqual({ processed: true });
  });

  it('should throw on invalid data', async () => {
    const job = new Job({
      queue: { name: 'test' },
      data: {},
      opts: { attempts: 1 },
    });
    await expect(processJob(job)).rejects.toThrow('Invalid data');
  });
});

Concurrency-Per-Worker Configuration

BullMQ workers accept a concurrency option that controls how many jobs a single worker processes in parallel. Default is 1. Increase concurrency for I/O-bound tasks (e.g., API calls, file reads). For CPU-bound tasks, keep concurrency low to avoid starving the event loop. Monitor CPU and memory usage to find the sweet spot. Use os.cpus().length as a starting point for CPU-bound tasks. For I/O-bound, you can go higher (e.g., 10-25). Each concurrent job uses a separate async context, so ensure your code is thread-safe (no shared mutable state).

const { Worker } = require('bullmq');
const os = require('os');

const worker = new Worker('myQueue', async job => {
  // I/O-bound task
  const result = await fetch(`https://api.example.com/data/${job.data.id}`);
  return result.json();
}, {
  connection,
  concurrency: 10, // process up to 10 jobs in parallel
});

// For CPU-bound, use os.cpus().length
const cpuWorker = new Worker('cpuQueue', async job => {
  // CPU-intensive work
  return heavyComputation(job.data);
}, {
  connection,
  concurrency: os.cpus().length,
});