Building a Redis Job Queue with BullMQ

BullMQ is the most popular Redis-based job queue for Node.js, providing reliable background job processing with features like delayed jobs, rate limiting, prioritization, and automatic retries. This guide covers setting up BullMQ for production workloads including common patterns and best practices.

Why BullMQ?

BullMQ (the successor to Bull) offers enterprise-grade job processing features:

• Guaranteed delivery — jobs are processed at least once, with configurable retry strategies
• Rate limiting — control job processing speed per queue or globally
• Delayed jobs — schedule jobs for future execution
• Prioritization — process high-priority jobs first
• Concurrency control — limit parallel job processing per worker
• Job dependencies — parent-child job flows
• Repeatable jobs — cron-like scheduled recurring jobs

Installation and Setup

npm install bullmq ioredis

# For the dashboard (optional)
npm install @bull-board/express @bull-board/api

Basic Queue and Worker

// queue.js — Producer
import { Queue } from 'bullmq';

const emailQueue = new Queue('email', {
  connection: {
    host: '127.0.0.1',
    port: 6379,
    maxRetriesPerRequest: null
  },
  defaultJobOptions: {
    attempts: 3,
    backoff: {
      type: 'exponential',
      delay: 2000,
    },
    removeOnComplete: { count: 1000 },
    removeOnFail: { count: 5000 },
  }
});

// Add a job
await emailQueue.add('welcome-email', {
  to: 'user@example.com',
  subject: 'Welcome!',
  template: 'welcome',
  vars: { name: 'John' }
});

// Add a delayed job (send in 1 hour)
await emailQueue.add('followup-email', {
  to: 'user@example.com',
  subject: 'How are things going?',
  template: 'followup'
}, {
  delay: 60 * 60 * 1000
});

// Add a priority job (lower number = higher priority)
await emailQueue.add('password-reset', {
  to: 'user@example.com',
  subject: 'Password Reset',
  template: 'reset'
}, {
  priority: 1
});

// worker.js — Consumer
import { Worker } from 'bullmq';

const worker = new Worker('email', async (job) => {
  console.log(`Processing ${job.name} for ${job.data.to}`);

  // Update progress
  await job.updateProgress(10);

  const html = await renderTemplate(job.data.template, job.data.vars);
  await job.updateProgress(50);

  await sendEmail(job.data.to, job.data.subject, html);
  await job.updateProgress(100);

  return { sent: true, messageId: 'abc123' };
}, {
  connection: { host: '127.0.0.1', port: 6379 },
  concurrency: 5,        // Process 5 jobs in parallel
  limiter: {
    max: 100,             // Max 100 jobs
    duration: 60000,      // Per minute
  }
});

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

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

Repeatable (Cron) Jobs

// Schedule recurring jobs
await emailQueue.add('daily-digest', { type: 'digest' }, {
  repeat: {
    pattern: '0 9 * * *',  // Every day at 9 AM
    tz: 'America/New_York'
  }
});

await emailQueue.add('weekly-report', { type: 'report' }, {
  repeat: {
    pattern: '0 8 * * MON',  // Every Monday at 8 AM
  }
});

// List repeatable jobs
const repeatableJobs = await emailQueue.getRepeatableJobs();
console.log(repeatableJobs);

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

Job Flows (Parent-Child Dependencies)

import { FlowProducer } from 'bullmq';

const flow = new FlowProducer({ connection: { host: '127.0.0.1' } });

// Parent job waits for all children to complete
await flow.add({
  name: 'generate-report',
  queueName: 'reports',
  data: { reportId: 123 },
  children: [
    {
      name: 'fetch-sales-data',
      queueName: 'data-fetch',
      data: { source: 'sales', reportId: 123 }
    },
    {
      name: 'fetch-user-data',
      queueName: 'data-fetch',
      data: { source: 'users', reportId: 123 }
    },
    {
      name: 'fetch-analytics',
      queueName: 'data-fetch',
      data: { source: 'analytics', reportId: 123 }
    }
  ]
});

Dashboard Setup

import express from 'express';
import { createBullBoard } from '@bull-board/api';
import { BullMQAdapter } from '@bull-board/api/bullMQAdapter';
import { ExpressAdapter } from '@bull-board/express';

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

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

const app = express();
app.use('/admin/queues', serverAdapter.getRouter());
app.listen(3000);

Error Handling and Retry Strategies

// Custom backoff strategy
await emailQueue.add('critical-email', data, {
  attempts: 5,
  backoff: {
    type: 'custom',
  },
});

// In worker, implement custom backoff
const worker = new Worker('email', processor, {
  settings: {
    backoffStrategy: (attemptsMade) => {
      // Fibonacci backoff: 1s, 1s, 2s, 3s, 5s
      const fib = [1000, 1000, 2000, 3000, 5000];
      return fib[Math.min(attemptsMade - 1, fib.length - 1)];
    }
  }
});

Graceful Shutdown

process.on('SIGTERM', async () => {
  console.log('Shutting down worker gracefully...');
  await worker.close();  // Waits for current jobs to finish
  process.exit(0);
});

Production Best Practices

• Always set removeOnComplete and removeOnFail with count limits to prevent Redis memory growth
• Use separate Redis instances for caching and job queues — queue operations should not compete with cache eviction
• Implement graceful shutdown to avoid lost jobs during deployments
• Monitor queue depth, processing time, and failure rate with Bull Board or Prometheus metrics
• Use maxRetriesPerRequest: null in the Redis connection to prevent ioredis from throwing on reconnection
• Set appropriate concurrency based on your job type — CPU-bound jobs should match core count, I/O-bound jobs can be higher
• Enable Redis persistence (RDB + AOF) for job durability across Redis restarts