Job Queues: Background Task Processing with Bull Redis Workers

Problem

Modern applications need to handle long-running operations without blocking user requests. Sending emails, processing images, generating reports, or syncing data shouldn't delay API responses. Traditional synchronous processing creates bottlenecks, poor user experience, and system instability.

Key Challenges:

• Long operations block request handlers
• No retry mechanism for failed tasks
• Difficult to track job status
• No priority handling
• Resource exhaustion from concurrent operations
• Lost jobs on server crashes

Solution

Bull is a Node.js library that provides a robust job queue system using Redis as the backing store. It enables asynchronous task processing with:

• Persistent job storage in Redis
• Automatic retries with exponential backoff
• Job prioritization and scheduling
• Progress tracking and status monitoring
• Concurrency control and rate limiting
• Dead letter queues for failed jobs
• Event-driven architecture with listeners

Code Implementation

1. Basic Setup

// npm install bull redis

const Queue = require('bull');
const redis = require('redis');

// Create a queue
const emailQueue = new Queue('emails', {
  redis: {
    host: 'localhost',
    port: 6379
  }
});

// Alternative: Using Redis connection
const redisClient = redis.createClient();
const taskQueue = new Queue('tasks', redisClient);

module.exports = emailQueue;

2. Producer: Adding Jobs

const emailQueue = require('./queue');

// Simple job
async function sendWelcomeEmail(userId, email) {
  const job = await emailQueue.add(
    { userId, email },
    {
      attempts: 3,
      backoff: {
        type: 'exponential',
        delay: 2000
      },
      removeOnComplete: true
    }
  );

  console.log(`Job ${job.id} added to queue`);
  return job.id;
}

// Delayed job (send after 1 hour)
async function scheduleReminder(userId, message) {
  await emailQueue.add(
    { userId, message },
    {
      delay: 3600000, // 1 hour
      attempts: 2
    }
  );
}

// Prioritized job
async function sendUrgentNotification(userId, message) {
  await emailQueue.add(
    { userId, message },
    {
      priority: 1, // Higher priority = processed first
      attempts: 5
    }
  );
}

// Batch jobs
async function sendBulkEmails(recipients) {
  const jobs = recipients.map(recipient =>
    emailQueue.add(recipient, { attempts: 3 })
  );

  return Promise.all(jobs);
}

module.exports = {
  sendWelcomeEmail,
  scheduleReminder,
  sendUrgentNotification,
  sendBulkEmails
};

3. Consumer: Processing Jobs

const emailQueue = require('./queue');
const nodemailer = require('nodemailer');

// Configure email service
const transporter = nodemailer.createTransport({
  service: 'gmail',
  auth: {
    user: process.env.EMAIL_USER,
    pass: process.env.EMAIL_PASS
  }
});

// Process jobs
emailQueue.process(5, async (job) => {
  console.log(`Processing job ${job.id}:`, job.data);

  try {
    const { userId, email, message } = job.data;

    // Update progress
    job.progress(25);

    // Validate email
    if (!email || !email.includes('@')) {
      throw new Error('Invalid email address');
    }

    job.progress(50);

    // Send email
    await transporter.sendMail({
      from: process.env.EMAIL_USER,
      to: email,
      subject: 'Welcome!',
      html: `<h1>Hello ${userId}</h1><p>${message}</p>`
    });

    job.progress(100);

    return { success: true, userId, email };
  } catch (error) {
    console.error(`Job ${job.id} failed:`, error.message);
    throw error; // Bull will retry
  }
});

// Event listeners
emailQueue.on('completed', (job, result) => {
  console.log(`✓ Job ${job.id} completed:`, result);
});

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

emailQueue.on('progress', (job, progress) => {
  console.log(`Job ${job.id} progress: ${progress}%`);
});

emailQueue.on('stalled', (job) => {
  console.warn(`Job ${job.id} stalled`);
});

module.exports = emailQueue;

4. Advanced: Multiple Queues

const Queue = require('bull');

// Different queues for different tasks
const emailQueue = new Queue('emails');
const imageQueue = new Queue('images');
const reportQueue = new Queue('reports');

// Email processor
emailQueue.process(10, async (job) => {
  // Send email logic
  return { sent: true };
});

// Image processor (fewer concurrent jobs)
imageQueue.process(2, async (job) => {
  const { imageUrl, format } = job.data;
  // Process image
  return { processed: true };
});

// Report processor (scheduled)
reportQueue.process(1, async (job) => {
  const { reportType, userId } = job.data;
  // Generate report
  return { reportId: 'report-123' };
});

// Add jobs from API
const express = require('express');
const app = express();

app.post('/api/send-email', async (req, res) => {
  const { email, subject } = req.body;
  const jobId = await emailQueue.add({ email, subject });
  res.json({ jobId, status: 'queued' });
});

app.post('/api/process-image', async (req, res) => {
  const { imageUrl, format } = req.body;
  const jobId = await imageQueue.add({ imageUrl, format });
  res.json({ jobId, status: 'queued' });
});

app.get('/api/job/:id', async (req, res) => {
  const job = await emailQueue.getJob(req.params.id);
  if (!job) return res.status(404).json({ error: 'Job not found' });

  const state = await job.getState();
  const progress = job._progress;

  res.json({ id: job.id, state, progress });
});

app.listen(3000);

5. Error Handling & Dead Letter Queue

const Queue = require('bull');

const taskQueue = new Queue('tasks');
const deadLetterQueue = new Queue('dead-letters');

taskQueue.process(async (job) => {
  const { retryCount = 0 } = job.data;

  try {
    // Simulate failure
    if (Math.random() < 0.3) {
      throw new Error('Random failure');
    }
    return { success: true };
  } catch (error) {
    // Move to dead letter after max retries
    if (job.attemptsMade >= job.opts.attempts) {
      await deadLetterQueue.add({
        originalJob: job.data,
        error: error.message,
        attempts: job.attemptsMade,
        timestamp: new Date()
      });
    }
    throw error;
  }
});

// Monitor dead letter queue
deadLetterQueue.process(async (job) => {
  console.error('Dead letter:', job.data);
  // Send alert, log to database, etc.
});

// Retry dead letter jobs manually
async function retryDeadLetterJob(jobId) {
  const job = await deadLetterQueue.getJob(jobId);
  if (job) {
    await taskQueue.add(job.data.originalJob, { attempts: 3 });
    await job.remove();
  }
}

module.exports = { taskQueue, deadLetterQueue, retryDeadLetterJob };

6. Monitoring & Dashboard

const Queue = require('bull');
const express = require('express');
const { createBullBoard } = require('@bull-board/express');
const { BullAdapter } = require('@bull-board/api/bullAdapter');

const app = express();

// Create queues
const emailQueue = new Queue('emails');
const imageQueue = new Queue('images');
const reportQueue = new Queue('reports');

// Setup Bull Board
const serverAdapter = new express.Router();
createBullBoard({
  queues: [
    new BullAdapter(emailQueue),
    new BullAdapter(imageQueue),
    new BullAdapter(reportQueue)
  ],
  serverAdapter
});

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

// Custom monitoring endpoint
app.get('/api/queue-stats', async (req, res) => {
  const stats = await Promise.all([
    emailQueue.getJobCounts(),
    imageQueue.getJobCounts(),
    reportQueue.getJobCounts()
  ]);

  res.json({
    email: stats[0],
    image: stats[1],
    report: stats[2]
  });
});

app.listen(3000);

Tips & Best Practices

1. Job Design

// ✓ Good: Idempotent, small payload
await queue.add({ userId: 123, action: 'send-email' });

// ✗ Bad: Large data, non-idempotent
await queue.add({ largeFile: buffer, timestamp: Date.now() });

2. Concurrency Control

// Process 5 jobs concurrently
queue.process(5, handler);

// Or dynamic based on load
queue.process(async (job) => {
  // Handler
});

3. Timeout Configuration

const job = await queue.add(data, {
  timeout: 30000 // 30 seconds max
});

4. Graceful Shutdown

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

5. Testing

// Use bull-testing for unit tests
const { mockQueue } = require('bull-testing');

const queue = mockQueue();
await queue.add({ test: true });
const jobs = await queue.getJobs();

6. Production Checklist

• Use Redis persistence (AOF/RDB)
• Monitor queue depth and processing time
• Set appropriate retry strategies
• Implement circuit breakers for external APIs
• Use separate Redis instances for different queues
• Enable Redis clustering for high availability
• Log all job failures
• Set up alerts for stalled jobs

Bull Redis workers provide enterprise-grade job processing for Node.js applications, enabling scalable, reliable background task handling.