Message Queues

Node.js Advanced — Message Queues

Node.js Advanced

Chapter 6 of 8 · Message Queues

Message Queues

A message queue is a buffer that sits between a producer (the thing that generates work) and a consumer (the thing that does the work). The producer doesn't wait for the work to finish — it drops a message in the queue and moves on. The consumer picks it up whenever it's ready. This decoupling solves three problems at once: bursts of load can be absorbed without dropping requests, slow consumers don't block fast producers, and failed jobs can be retried automatically without any code in the original request path. In Node, the dominant queue stack is Redis-backed BullMQ — fast, durable, and richly featured.

Why Queue Instead of Awaiting Directly?

Scenario	Direct `await`	Message queue
Send a welcome email on signup	User waits for email server; if it's down, signup fails	Signup completes instantly; email sent asynchronously with retries
Resize 50 uploaded images	Request handler blocks for seconds; client times out	Jobs queued immediately; workers process at their own pace
Traffic spike: 10,000 orders/minute	DB connection pool exhausted; requests fail	Queue absorbs the burst; workers drain at a controlled rate
External API call fails mid-processing	Work is lost; user must retry manually	Job stays in queue and retries with backoff automatically

Core Vocabulary

Producer

The code that adds a job to the queue. Usually a request handler or another job. It doesn't know or care how many workers are consuming.

Consumer / Worker

The code that picks jobs off the queue and processes them. Can run in a separate process, on separate machines, or in parallel with many workers on the same queue.

Acknowledgement (ACK)

The worker tells the broker "I finished this job." Until the broker receives an ACK, it keeps the job in an "active" state so it can re-queue it if the worker crashes mid-job.

Dead Letter Queue (DLQ)

Where jobs land after exhausting all retries. They aren't deleted — they're parked so you can inspect them, fix the bug, and re-process them manually.

Pub/Sub

Fan-out pattern: one publisher sends a message to a channel; all subscribers receive a copy. Fire-and-forget — if a subscriber is offline, it misses the message.

Competing Consumers

Multiple workers reading from the same queue. Each job is delivered to exactly one worker. Scales horizontally — add workers to increase throughput.

Redis Pub/Sub — Lightweight Fan-Out

Redis Pub/Sub is the simplest messaging primitive: a publisher pushes a message to a named channel; every current subscriber receives it. There is no persistence — messages sent when no subscriber is listening are lost.

// publisher.js
const Redis = require('ioredis');
const pub   = new Redis();

async function publishEvent(channel, payload) {
    await pub.publish(channel, JSON.stringify(payload));
}

await publishEvent('order.created', { orderId: 123, userId: 42 });
    

// subscriber.js — a subscribed connection CANNOT send other commands
const Redis = require('ioredis');
const sub   = new Redis();   // separate connection from the publisher

sub.subscribe('order.created', 'order.updated');

sub.on('message', (channel, message) => {
    const payload = JSON.parse(message);
    console.log(`[${channel}]`, payload);
});

// Pattern subscriptions — e.g. all order.* channels
sub.psubscribe('order.*');
sub.on('pmessage', (pattern, channel, message) => {
    console.log(`[${pattern} → ${channel}]`, JSON.parse(message));
});
    

⚠️ Redis Pub/Sub Limitations

No persistence. If a subscriber is down or slow, messages are lost — there's no buffer. Use Redis Streams or BullMQ if you need durability or retry.

No acknowledgement. The publisher has no way to know if any subscriber received or processed the message.

Best for: live WebSocket broadcasts (broadcast to all connected workers to push to their sockets), cache invalidation signals, low-stakes notifications where the occasional dropped message is acceptable.

BullMQ — Durable Job Queues on Redis

BullMQ is the standard Node.js job queue library. It uses Redis for persistence, supports concurrent workers, retries with configurable backoff, delayed jobs, repeating (cron) jobs, prioritisation, and job progress reporting.

// npm install bullmq ioredis

// producer.js — add jobs to the queue
const { Queue } = require('bullmq');

const emailQueue = new Queue('email', {
    connection: { host: 'localhost', port: 6379 },
    defaultJobOptions: {
        attempts:  3,                        // retry up to 3 times
        backoff:   { type: 'exponential', delay: 2000 },  // 2s, 4s, 8s
        removeOnComplete: { count: 1000 },  // keep last 1000 completed jobs
        removeOnFail:     { count: 500  },  // keep last 500 failed jobs for inspection
    },
});

// Add a regular job
await emailQueue.add('welcome', { to: 'alice@example.com', name: 'Alice' });

// Add a delayed job — send after 1 hour
await emailQueue.add('reminder', { to: 'alice@example.com' }, { delay: 60 * 60 * 1000 });

// Add a high-priority job (lower number = higher priority)
await emailQueue.add('alert', { to: 'admin@example.com' }, { priority: 1 });

// Repeating job — runs every day at 09:00 UTC
await emailQueue.add(
    'daily-digest',
    { type: 'digest' },
    { repeat: { pattern: '0 9 * * *' } },
);
    

// worker.js — process jobs
const { Worker } = require('bullmq');

const worker = new Worker(
    'email',
    async (job) => {
        // job.name    — the job type string ('welcome', 'reminder', ...)
        // job.data    — the payload passed to queue.add()
        // job.id      — unique job ID
        // job.attemptsMade — how many times this job has been attempted

        switch (job.name) {
            case 'welcome':
                await sendWelcomeEmail(job.data);
                break;
            case 'reminder':
                await sendReminderEmail(job.data);
                break;
            default:
                throw new Error(`Unknown job type: ${job.name}`);
        }

        // Return value is stored on job.returnvalue
        return { sentAt: new Date().toISOString() };
    },
    {
        connection:  { host: 'localhost', port: 6379 },
        concurrency: 5,       // process up to 5 jobs simultaneously
    },
);

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

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

Job Lifecycle

┌─────────────┐ queue.add() ───►│ waiting │ └──────┬──────┘ │ worker picks up ┌──────▼──────┐ │ active │ ◄── job.updateProgress() └──────┬──────┘ ┌──────────────┼──────────────┐ throws │ │ returns │ delay passed ▼ ▼ ▼ ┌──────────┐ ┌───────────┐ ┌──────────────┐ │ waiting │ │ completed │ │ delayed │ │ (retry) │ └───────────┘ └──────┬───────┘ └──────┬───┘ │ max attempts │ back to waiting reached│ ▼ ┌─────────────┐ │ failed │ ◄── inspect / re-queue manually └─────────────┘

Job Progress Reporting

// Inside the worker processor — report progress (0–100)
async (job) => {
    const items = job.data.items;
    for (let i = 0; i < items.length; i++) {
        await processItem(items[i]);
        await job.updateProgress(Math.round(((i + 1) / items.length) * 100));
    }
}

// Observe progress from outside (e.g. a status API)
const { QueueEvents } = require('bullmq');
const queueEvents = new QueueEvents('email', { connection: { host: 'localhost', port: 6379 } });

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

// Wait for a specific job to complete (useful in tests or sync-like flows)
const job = await emailQueue.add('welcome', { to: 'bob@example.com' });
const result = await job.waitUntilFinished(queueEvents, 30_000);
    

Rate Limiting Workers

// Limit how fast workers consume — e.g. to respect an external API rate limit
const worker = new Worker('email', processor, {
    connection: { host: 'localhost', port: 6379 },
    limiter: {
        max:      10,     // process at most 10 jobs
        duration: 1000,  // per 1000ms (1 second)
    },
});
    

Inspecting the Queue

const { Queue } = require('bullmq');
const q = new Queue('email', { connection: { host: 'localhost', port: 6379 } });

// Counts in each state
const counts = await q.getJobCounts('waiting', 'active', 'completed', 'failed', 'delayed');
console.log(counts);  // { waiting: 42, active: 5, completed: 1200, failed: 3, delayed: 10 }

// Fetch failed jobs to inspect
const failed = await q.getFailed(0, 9);   // first 10
for (const job of failed) {
    console.log(job.id, job.data, job.failedReason);
}

// Retry all failed jobs
const failedJobs = await q.getFailed(0, 99);
await Promise.all(failedJobs.map(j => j.retry()));

// Drain the queue (remove all waiting jobs)
await q.drain();

// Obliterate — deletes everything including Redis keys (dev/testing only)
await q.obliterate({ force: true });
    

Flow Producers — Job Dependencies

A Flow lets you define parent/child job relationships. A parent job only moves to active after all its children complete. Useful for pipelines where you need all sub-tasks to finish before the next step runs.

const { FlowProducer } = require('bullmq');
const flow = new FlowProducer({ connection: { host: 'localhost', port: 6379 } });

// 'notify-user' runs only after both 'resize-image' and 'update-db' complete
await flow.add({
    name:    'notify-user',
    queueName: 'notifications',
    data:    { userId: 42 },
    children: [
        { name: 'resize-image', queueName: 'images', data: { src: 'upload.jpg', width: 800 } },
        { name: 'update-db',     queueName: 'db',     data: { userId: 42, status: 'processed' } },
    ],
});
    

Graceful Shutdown with BullMQ

// Give in-flight jobs time to finish before the process exits
async function shutdown() {
    console.log('Shutting down workers...');

    // close() waits for the current job to finish before stopping
    await worker.close();

    // Close the queue connection too
    await emailQueue.close();

    console.log('Shutdown complete');
    process.exit(0);
}

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

💡 BullMQ vs Bull (v3)

BullMQ is the rewrite of the original Bull library. Key differences: BullMQ requires ioredis 5+ and uses a dedicated Worker class (Bull used the Queue for both producing and consuming). BullMQ is the actively maintained library — new projects should use BullMQ; Bull receives security fixes only.

For a web dashboard to inspect queues visually: Bull Board (@bull-board/express) works with both. Mount it at /admin/queues and protect it behind auth middleware.

Feature	Redis Pub/Sub	Redis Streams	BullMQ
Persistence	No — missed if offline	Yes — replay from any position	Yes — stored in Redis until ACK'd
Delivery	Fan-out (all subscribers)	Fan-out or competing consumers	Competing consumers (one worker per job)
Retries	None	Manual via PEL	Built-in with backoff, max attempts
Delayed/Scheduled	No	No	Yes — delay, cron, repeat
Complexity	Low	Medium	Medium — but rich built-in features
Best for	Live events, cache invalidation	Event sourcing, audit logs	Background jobs, email, image processing

Coding Challenges

Challenge 1 — In-Memory Job Queue

Build a SimpleQueue class (no Redis, no external packages) that: enqueues jobs with add(name, data, { priority, delayMs }), dispatches them to a registered processor function with process(name, concurrency, fn), retries failed jobs up to a configurable maxAttempts with a fixed delay between attempts, and emits 'completed' and 'failed' events. Demonstrate it with a 10-job batch where 3 jobs intentionally fail twice before succeeding on the third attempt, and verify the event counts are correct.

View sample solution ↗

Challenge 2 — BullMQ Email Worker

Build a complete BullMQ producer/worker pair for an email queue. The producer exposes enqueueWelcome(to, name), enqueueReminder(to, delayMs), and enqueueDigest() (repeating, daily at 09:00). The worker handles all three job types, reports progress at 50% and 100%, and on failure logs the job ID, attempt number, and error. Write a test that enqueues 5 welcome jobs with a mock sender, waits for all to complete using job.waitUntilFinished, and asserts the mock was called exactly 5 times with the right data. Use ioredis-mock for the Redis connection.

View sample solution ↗

Challenge 3 — Queue Health Dashboard

Build an Express route GET /queue-health that returns a JSON snapshot of every named queue passed to it: job counts per state (waiting/active/completed/failed/delayed), the oldest failed job (id, data, failedReason, attemptsMade), and a simple health verdict ("healthy" if failed count is 0, "degraded" if 1–9, "critical" if 10+). Protect the route with a hard-coded API key check using crypto.timingSafeEqual. Write tests that mock the Queue instance and verify the health verdict logic for all three bands.

View sample solution ↗