Node.js에서 Bull 큐 작업 처리

Introduction

Background job processing is essential for operations that are too slow for HTTP request/response cycles: sending emails, processing images, generating reports, and more. Bull is a robust queue library built on Redis that handles job scheduling, retries, and rate limiting.

Environment

node --version
# v20.11.0

npm list bull
# my-app@1.0.0
# +-- bull@4.12.0

redis-cli ping
# PONG

Problem

Synchronous job processing blocks the event loop and causes timeouts:

app.post('/send-email', async (req, res) => {
  const { to, subject, body } = req.body;
  
  // This takes 5+ seconds and blocks other requests!
  await sendEmail(to, subject, body);
  
  res.json({ message: 'Email sent' });
});

Analysis

Bull queue architecture:

Producer → Redis Queue → Worker → Redis → Completion
                ↓
            Delayed jobs
            Retries
            Rate limiting
            Priority

The key components:

1. Queue - Named job container
2. Producer - Adds jobs to the queue
3. Worker - Processes jobs from the queue
4. Redis - Stores job data and state

Solution

Solution 1: Basic queue setup

// queue.js
const Queue = require('bull');

const emailQueue = new Queue('email', {
  redis: {
    host: '127.0.0.1',
    port: 6379
  }
});

module.exports = emailQueue;

// Producer (in route handler)
app.post('/send-email', async (req, res) => {
  const { to, subject, body } = req.body;
  
  await emailQueue.add('send', {
    to,
    subject,
    body
  }, {
    attempts: 3,
    backoff: {
      type: 'exponential',
      delay: 2000
    }
  });
  
  res.json({ message: 'Email queued' });
});

// Worker (separate process)
emailQueue.process('send', async (job) => {
  const { to, subject, body } = job.data;
  console.log(`Sending email to ${to}`);
  
  // Simulate email sending
  await sendEmail(to, subject, body);
  
  return { success: true };
});

Solution 2: Multiple job types

const Queue = require('bull');

const imageQueue = new Queue('image-processing');
const reportQueue = new Queue('report-generation');

// Image processing jobs
imageQueue.process('resize', 3, async (job) => {
  const { imagePath, sizes } = job.data;
  
  for (const size of sizes) {
    await resizeImage(imagePath, size);
    await job.progress(Math.random() * 100);
  }
  
  return { processed: sizes.length };
});

imageQueue.process('watermark', async (job) => {
  const { imagePath, text } = job.data;
  await addWatermark(imagePath, text);
});

// Report generation jobs
reportQueue.process('generate', async (job) => {
  const { userId, dateRange } = job.data;
  const report = await generateReport(userId, dateRange);
  return { reportUrl: report.url };
});

// Add jobs
app.post('/api/images/resize', async (req, res) => {
  const job = await imageQueue.add('resize', {
    imagePath: req.body.path,
    sizes: ['thumb', 'medium', 'large']
  });
  
  res.json({ jobId: job.id });
});

app.post('/api/reports/generate', async (req, res) => {
  const job = await reportQueue.add('generate', {
    userId: req.user.id,
    dateRange: req.body
  }, {
    priority: 1,
    removeOnComplete: 100
  });
  
  res.json({ jobId: job.id });
});

Solution 3: Job progress and events

const Queue = require('bull');

const processQueue = new Queue('data-processing');

// Track progress
processQueue.process(async (job) => {
  const totalSteps = 10;
  
  for (let i = 0; i < totalSteps; i++) {
    // Do work
    await processChunk(job.data.chunks[i]);
    
    // Update progress
    await job.progress(((i + 1) / totalSteps) * 100);
  }
  
  return { completed: true };
});

// Listen to events
processQueue.on('progress', (job, progress) => {
  console.log(`Job ${job.id}: ${progress}% complete`);
});

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

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

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

Solution 4: Delayed and scheduled jobs

const Queue = require('bull');

const reminderQueue = new Queue('reminders');

// Add delayed job (execute after delay)
await reminderQueue.add('send', {
  userId: user.id,
  message: 'Your report is ready'
}, {
  delay: 5000, // 5 seconds
  attempts: 3
});

// Add scheduled job (execute at specific time)
const scheduledDate = new Date('2024-01-15T09:00:00Z');
await reminderQueue.add('send', {
  userId: user.id,
  message: 'Morning report'
}, {
  delay: scheduledDate.getTime() - Date.now(),
  repeat: {
    cron: '0 9 * * *' // Every day at 9 AM
  }
});

Solution 5: Rate limiting

const Queue = require('bull');

const emailQueue = new Queue('emails', {
  redis: { host: '127.0.0.1', port: 6379 },
  limiter: {
    max: 100,        // Max jobs per interval
    duration: 60000  // Per 60 seconds
  }
});

// This ensures we send max 100 emails per minute
emailQueue.process(async (job) => {
  await sendEmail(job.data.to, job.data.subject, job.data.body);
});

Lessons Learned

1. Use separate queues for different job types - It makes monitoring easier
2. Set retry limits - Infinite retries can cause resource exhaustion
3. Monitor queue metrics - Use Bull Board for visualization
4. Handle stalled jobs - Jobs that take too long should be retried
5. Clean completed jobs - Use removeOnComplete to prevent memory leaks

This blog does not accept any external sponsorships, affiliate marketing, or ad revenue.