Node.js 62 🟢 Asynchronous Background Task Processing with BullMQ and Redis - KandZ

Every web application eventually hits the same wall: some operations are too slow to run inside an HTTP request. Sending an email takes seconds. Generating a PDF report takes minutes. Resizing a video takes hours. If you run these tasks synchronously, the user waits, the connection times out, and the server becomes unresponsive under load. The solution is to decouple the work from the request: enqueue it, return immediately, and let a separate process handle it in the background.

BullMQ is the standard Node.js library for this pattern. It is a Redis-based job queue that stores jobs, manages their lifecycle, and distributes them to workers . The architecture is simple: a Queue accepts jobs from your API, and a Worker processes them in a separate process. Redis sits in the middle as the durable store, holding job data, state, and metadata . This chapter covers the core patterns: defining queues, adding jobs with retries and backoff, processing with workers and concurrency, handling failures, and monitoring with events.

Key point: BullMQ separates the enqueueing of work (fast, synchronous, returns immediately) from the processing of work (slow, asynchronous, runs in separate worker processes), using Redis as the durable intermediate store.

Why BullMQ exists

The request-response mismatch. An HTTP request has a deadline—often 30 seconds, sometimes less. A background task does not. Image processing, video encoding, report generation, and third-party API calls can take arbitrary time. BullMQ lets the request return 202 Accepted immediately while the work continues independently .

The reliability problem. If you use setTimeout or an in-process queue, the work is lost when the process restarts. BullMQ stores every job in Redis, so a crash does not lose work. Workers automatically recover stalled jobs—jobs whose worker died mid-processing—and retry them .

The scaling problem. A single Node.js process can only do so much. BullMQ distributes jobs across multiple worker processes, on the same machine or across many machines. You can scale processing horizontally by starting more workers, without changing any application code .

The failure handling problem. Network calls fail. APIs return errors. Files are locked. BullMQ provides retries with exponential backoff, job timeouts, and failure tracking. A job that fails three times is marked failed, and you can inspect why .

The visibility problem. Without a queue, you have no idea how much work is pending, how long it is taking, or what is failing. BullMQ exposes events for every state transition: waiting, active, completed, failed, progress. You can build dashboards, send alerts, and trace every job’s lifecycle .

a. Queues and adding jobs

A Queue is the entry point. It is a lightweight object that connects to Redis and lets you add jobs. It does not process anything—that is the worker’s job.

import { Queue } from 'bullmq';
import IORedis from 'ioredis';

const connection = new IORedis(process.env.REDIS_URL, {
  maxRetriesPerRequest: null
});

const emailQueue = new Queue('emails', {
  connection,
  defaultJobOptions: {
    attempts: 3,
    backoff: {
      type: 'exponential',
      delay: 1000
    },
    removeOnComplete: { count: 1000 },
    removeOnFail: { count: 5000 }
  }
});

The maxRetriesPerRequest: null setting is required for BullMQ. Without it, ioredis gives up after a number of retries, but BullMQ expects to retry forever . Each BullMQ component—Queue, Worker, QueueEvents—needs its own Redis connection, because BullMQ uses blocking commands that would stall a shared connection .

Adding a job is fast. It writes to Redis and returns a Job object:

await emailQueue.add('sendWelcome', {
  to: 'user@example.com',
  subject: 'Welcome',
  templateId: 'welcome',
  variables: { name: 'Alice' }
});

The first argument is the job name, which you can use to route different job types to the same queue. The second is the payload—any JSON-serializable data. BullMQ stores the job metadata in Redis Hashes and the job state as membership in Redis data structures (wait, active, completed, failed) .

You can add jobs in bulk with addBulk, which reduces Redis round-trips:

await emailQueue.addBulk([
  { name: 'sendWelcome', data: { to: 'a@example.com' } },
  { name: 'sendWelcome', data: { to: 'b@example.com' } }
]);

Jobs can be delayed, scheduled, or prioritized. A delayed job waits before becoming eligible for processing:

await emailQueue.add('reminder', data, { delay: 60000 });

b. Workers and concurrency

A Worker is the processing engine. It connects to the same Redis instance, listens on the queue, and runs your processor function for each job.

import { Worker } from 'bullmq';

const worker = new Worker(
  'emails',
  async (job) => {
    const { to, subject, templateId, variables } = job.data;

    await job.updateProgress(10);
    const html = await renderTemplate(templateId, variables);
    await job.updateProgress(50);

    const result = await sendEmail({ to, subject, html });
    await job.updateProgress(100);

    return { messageId: result.id, sentAt: new Date().toISOString() };
  },
  {
    connection,
    concurrency: 5
  }
);

The concurrency option controls how many jobs a single worker processes in parallel . It is only useful when the processor performs asynchronous operations—database calls, HTTP requests, file I/O. If the processor is CPU-bound, concurrency does not help because Node.js is single-threaded; use sandboxed processors for CPU-intensive work .

You can run multiple workers for the same queue, in the same process or different processes. BullMQ distributes jobs across them in a round-robin fashion . For production, run workers in separate processes from your API server, so a crashed worker does not take down the web server .

The processor can return a value, which becomes the job’s return value and is passed to completion events. It can report progress with job.updateProgress(), which emits progress events.

c. Events, retries, and graceful shutdown

BullMQ exposes events at two levels: on the worker for local processing events, and on a QueueEvents instance for global events across all workers.

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

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

For global monitoring, create a QueueEvents instance:

import { QueueEvents } from 'bullmq';

const queueEvents = new QueueEvents('emails', { connection });

queueEvents.on('waiting', ({ jobId }) => {
  console.log(`Job ${jobId} is waiting`);
});

queueEvents.on('progress', ({ jobId, data }) => {
  console.log(`Job ${jobId} is ${data}% complete`);
});

QueueEvents does not emit the Job instance for performance reasons—only the jobId. If you need the full job, call Job.fromId .

Retries are configured on the job or as defaults on the queue. The attempts option sets how many times a job is tried before being marked failed. The backoff option controls the delay between retries—exponential with an initial delay is the common pattern .

Graceful shutdown is essential in production. When your process receives SIGTERM (from Docker, Kubernetes, or a process manager), call worker.close():

process.on('SIGTERM', async () => {
  await worker.close();
  await emailQueue.close();
  process.exit(0);
});

worker.close() marks the worker as closing, stops accepting new jobs, and waits for active jobs to finish . If the shutdown is ungraceful, stalled jobs are detected and reassigned to other workers .

Complete Example Session

The following session builds a complete email queue with retries, progress reporting, and event monitoring.

// ============================================
// PART 1: IMPORTS AND CONNECTION
// ============================================
import { Queue, Worker, QueueEvents, Job } from 'bullmq';
import IORedis from 'ioredis';

const connection = new IORedis(process.env.REDIS_URL, {
  maxRetriesPerRequest: null
});

// ============================================
// PART 2: QUEUE WITH DEFAULT OPTIONS
// ============================================
const emailQueue = new Queue('emails', {
  connection: new IORedis(process.env.REDIS_URL, {
    maxRetriesPerRequest: null
  }),
  defaultJobOptions: {
    attempts: 3,
    backoff: { type: 'exponential', delay: 1000 },
    removeOnComplete: { count: 1000 },
    removeOnFail: { count: 5000 }
  }
});

// ============================================
// PART 3: ADDING JOBS
// ============================================
await emailQueue.add('sendWelcome', {
  to: 'alice@example.com',
  subject: 'Welcome',
  templateId: 'welcome',
  variables: { name: 'Alice' }
});

// ============================================
// PART 4: THE WORKER PROCESSOR
// ============================================
const worker = new Worker(
  'emails',
  async (job: Job) => {
    const { to, subject, templateId, variables } = job.data;

    await job.updateProgress(10);

    // Simulate template rendering
    const html = `<h1>${variables.name}</h1>`;
    await job.updateProgress(50);

    // Simulate sending email
    await new Promise(resolve => setTimeout(resolve, 100));
    await job.updateProgress(100);

    return {
      messageId: `msg_${job.id}`,
      sentAt: new Date().toISOString()
    };
  },
  {
    connection: new IORedis(process.env.REDIS_URL, {
      maxRetriesPerRequest: null
    }),
    concurrency: 5
  }
);

// ============================================
// PART 5: WORKER EVENT HANDLERS
// ============================================
worker.on('completed', (job, result) => {
  console.log(`Email to ${job.data.to} sent (${result.messageId})`);
});

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

// ============================================
// PART 6: GLOBAL EVENT MONITORING
// ============================================
const queueEvents = new QueueEvents('emails', {
  connection: new IORedis(process.env.REDIS_URL, {
    maxRetriesPerRequest: null
  })
});

queueEvents.on('waiting', ({ jobId }) => {
  console.log(`Job ${jobId} waiting`);
});

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

// ============================================
// PART 7: ADDING A FAILING JOB (for demonstration)
// ============================================
await emailQueue.add('sendWelcome', {
  to: 'fail@example.com',
  subject: 'This will fail',
  templateId: 'welcome',
  variables: { name: 'Fail' }
});

// ============================================
// PART 8: GRACEFUL SHUTDOWN
// ============================================
process.on('SIGTERM', async () => {
  await worker.close();
  await queueEvents.close();
  await emailQueue.close();
  process.exit(0);
});

// ============================================
// PART 9: SCHEDULED JOB
// ============================================
await emailQueue.add(
  'sendReminder',
  { to: 'alice@example.com', subject: 'Reminder' },
  { delay: 3600000 } // 1 hour
);

// ============================================
// PART 10: START
// ============================================
console.log('Worker started, waiting for jobs...');

The ten parts cover connection setup, queue definition with defaults, adding a job, the worker processor with progress reporting, worker event handlers, global QueueEvents monitoring, a demonstration of failure handling, graceful shutdown, a scheduled job, and startup. What the session demonstrates is the separation of concerns: the queue accepts jobs and returns immediately, the worker processes them asynchronously, and events provide visibility into every state transition.

Quick Reference

Core Classes

Class	Purpose
`Queue`	Add jobs, manage queue
`Worker`	Process jobs
`QueueEvents`	Global event monitoring
`FlowProducer`	Parent/child job flows

Job Options

Option	Default	Meaning
`attempts`	1	Retry attempts before failure
`backoff`	none	Delay strategy between retries
`delay`	0	Milliseconds before job becomes eligible
`priority`	0	Lower number = higher priority
`removeOnComplete`	false	Keep or remove completed jobs
`removeOnFail`	false	Keep or remove failed jobs

Worker Options

Option	Default	Meaning
`concurrency`	1	Parallel jobs per worker
`lockDuration`	30000	Job lock time in ms
`stalledInterval`	30000	Stalled check interval

Required Redis Setting

Setting	Value	Why
`maxRetriesPerRequest`	`null`	BullMQ requires infinite retry

Best Practices

✅ Do This:

new IORedis(url, { maxRetriesPerRequest: null });          // ✅ required
worker.on('failed', (job, err) => log(err));                // ✅ handle failures
await worker.close();                                       // ✅ graceful shutdown
attempts: 3, backoff: { type: 'exponential', delay: 1000 } // ✅ retry strategy

❌ Don’t Do This:

new IORedis();                                              // ❌ missing maxRetriesPerRequest
worker = new Worker('q', fn);                               // ❌ no error listener
process.exit(0);                                            // ❌ without worker.close()
concurrency: 1000;                                          // ❌ too high for CPU-bound work

Common Pitfalls

Pitfall	Why It Happens	Fix
“maxRetriesPerRequest must be null”	Missing Redis option	Set it to `null`
Jobs never processed	No worker running	Start a worker process
Worker stalls after Redis disconnect	Blocking connection issue	Separate connections per component
Memory grows unbounded	`removeOnComplete` not set	Set retention limits
Duplicate processing	Multiple workers, slow lock renewal	Increase lock duration or reduce concurrency
Jobs lost on restart	No Redis persistence	Enable Redis AOF or RDB

Real-World Examples

1. Email sending queue

await emailQueue.add('send', { to, subject, body });

2. Image processing

await imageQueue.add('resize', { path, width: 800 });

3. Report generation

await reportQueue.add('generate', { userId, format: 'pdf' });

4. Webhook delivery

await webhookQueue.add('deliver', { url, payload }, { attempts: 5 });

5. Scheduled reminder

await queue.add('remind', data, { delay: 86400000 });

6. Bulk enqueue

await queue.addBulk(items.map(i => ({ name: 'process', data: i })));

7. Priority job

await queue.add('urgent', data, { priority: 1 });

8. Progress reporting

await job.updateProgress(50);

9. Graceful shutdown

process.on('SIGTERM', async () => { await worker.close(); });

10. Global concurrency limit

await queue.setGlobalConcurrency(10);

Visual

BullMQ Architecture

┌──────────────────────────────────────────────┐
│  API SERVER                                  │
│  ┌────────────────────────────────────────┐  │
│  │  emailQueue.add('send', data)          │  │
│  │  returns immediately                   │  │
│  └────────────────┬───────────────────────┘  │
│                   │                          │
└───────────────────┼──────────────────────────┘
                    │
                    ▼
┌──────────────────────────────────────────────┐
│  REDIS                                       │
│  ┌────────────────────────────────────────┐  │
│  │  bull:emails:wait    (job IDs)         │  │
│  │  bull:emails:active  (processing)      │  │
│  │  bull:emails:completed                 │  │
│  │  bull:emails:failed                    │  │
│  │  bull:emails:*       (job hashes)      │  │
│  └────────────────────────────────────────┘  │
└───────────────────┬──────────────────────────┘
                    │
                    ▼
┌──────────────────────────────────────────────┐
│  WORKER PROCESS                              │
│  ┌────────────────────────────────────────┐  │
│  │  Worker('emails', async (job) => {     │  │
│  │    // process job.data                 │  │
│  │    return result;                      │  │
│  │  }, { concurrency: 5 })                │  │
│  └────────────────────────────────────────┘  │
└──────────────────────────────────────────────┘

Job Lifecycle

┌──────────────────────────────────────────────┐
│  add()                                       │
│    │                                         │
│    ▼                                         │
│  WAIT ──▶ ACTIVE ──▶ COMPLETED               │
│    │        │                                │
│    │        │ (error)                        │
│    │        ▼                                │
│    │      FAILED ──▶ (retry) ──▶ WAIT        │
│    │        │                                │
│    │        │ (attempts exhausted)           │
│    │        ▼                                │
│    │      FAILED (permanent)                 │
│    │                                         │
│    └──▶ DELAYED ──▶ WAIT (after delay)       │
└──────────────────────────────────────────────┘

Concurrency vs Multiple Workers

┌──────────────────────────────────────────────┐
│  SINGLE WORKER, concurrency: 3               │
│  ┌────────────────────────────────────────┐  │
│  │  Worker                                │  │
│  │  ┌─────┐ ┌─────┐ ┌─────┐               │  │
│  │  │Job 1│ │Job 2│ │Job 3│               │  │
│  │  └─────┘ └─────┘ └─────┘               │  │
│  └────────────────────────────────────────┘  │
│                                              │
│  MULTIPLE WORKERS, concurrency: 1 each       │
│  ┌──────────┐  ┌──────────┐  ┌──────────┐    │
│  │ Worker 1 │  │ Worker 2 │  │ Worker 3 │    │
│  │  Job 1   │  │  Job 2   │  │  Job 3   │    │
│  └──────────┘  └──────────┘  └──────────┘    │
│  Higher availability, same throughput        │
└──────────────────────────────────────────────┘

Summary

Item	Value
Queue	Adds jobs to Redis
Worker	Processes jobs
QueueEvents	Global event monitoring
Required Redis option	`maxRetriesPerRequest: null`
Default concurrency	1
Retry attempts	`attempts` option
Backoff	`backoff` option
Graceful shutdown	`await worker.close()`
Job storage	Redis Hashes + state lists
Atomicity	Lua scripts

Key takeaways:

BullMQ decouples enqueueing from processing. The API adds jobs and returns immediately; workers process asynchronously in separate processes .
Redis is the durable store. Jobs survive process restarts. Redis Hashes store job data; Redis lists and sets store state .
maxRetriesPerRequest: null is required. BullMQ uses blocking Redis commands and expects infinite retry .
Concurrency controls parallel jobs per worker. It is useful for I/O-bound work, not CPU-bound work .
Retries and backoff are built in. Configure attempts and backoff per job or as queue defaults .
Graceful shutdown prevents stalled jobs. Call worker.close() on SIGTERM to wait for active jobs .
Events provide visibility. Worker events are local; QueueEvents are global across all workers .

Remember: BullMQ is the standard solution for background job processing in Node.js. The pattern is always the same: define a queue, add jobs with options for retries and delays, run workers to process them, and listen to events for monitoring. Redis stores everything, so jobs survive restarts. Workers can run anywhere—same machine, different machine, many machines—and BullMQ distributes work across them. Use it for emails, reports, image processing, webhooks, and any task that should not block an HTTP request.

Stop using slow, ad-bloated tool sites! 🤮

🔎 Search “KandZ Tools” on Google to use many professional utilities for free.

KandZ.me is the ultimate minimalist hub for:
✅ Finance (Mortgage, Interest, Inflation)
✅ Tech (Base64, JSON, Dev Suite, IP)
✅ Health (BMI, BMR, TDEE)
✅ Productivity (Timer, Workspace, QR)

⚡️ Fast & Private
🔒 No data leaves your device
💎 100% Free

🔗 Use it now: https://tools.kandz.me
🔖 Bookmark it—you’ll need it later!