Node.js 비동기 작업을 안정적으로 처리하는 BullMQ 5 잡 큐 — 재시도 전략·Dead Letter Queue·Sandboxed Processor·Prometheus 메트릭 연동까지

npm install bullmq ioredis

// src/queue/config.ts
import IORedis from 'ioredis';
 
const connection = new IORedis({
  host: process.env.REDIS_HOST || 'localhost',
  port: 6379,
  maxRetriesPerRequest: null, // BullMQ 필수 설정
});
 
export { connection };

import { Worker, Job } from 'bullmq';
import { connection } from './config';
 
const worker = new Worker(
  'email',
  async (job) => { /* 처리 로직 */ },
  {
    connection,
    concurrency: 10,
    settings: {
      // HTTP 429 응답의 Retry-After 헤더 값을 우선 사용하는 예시
      backoffStrategy: (attemptsMade: number, type: string, err: any, job: Job) => {
        if (err?.retryAfter) {
          return err.retryAfter * 1000;
        }
        return Math.pow(2, attemptsMade) * 1000;
      },
    },
  }
);
 
// 잡 추가 시 'custom' 타입으로 위 함수를 참조
await emailQueue.add('jobName', data, {
  attempts: 5,
  backoff: { type: 'custom' },
});

// src/queue/email.queue.ts
import { Queue } from 'bullmq';
import { connection } from './config';
 
const emailQueue = new Queue('email', { connection });
 
await emailQueue.add(
  'sendWelcomeEmail',
  { userId: 'user-123', email: 'user@example.com', name: '홍길동' },
  {
    attempts: 5,
    backoff: {
      type: 'exponential',
      delay: 3000, // 1차 재시도: 3s, 2차: 6s, 3차: 12s, 4차: 24s
      jitter: 0.5, // delay의 50%~100% 범위로 무작위 편차 적용해 Thundering Herd 방지
    },
    removeOnComplete: 100, // 완료된 잡 최대 100개만 보관
    removeOnFail: 500,     // 실패한 잡 최대 500개만 보관
  }
);

// src/queue/email.worker.ts
import { Worker, Job, MetricsTime } from 'bullmq';
import { connection } from './config';
import sgMail from '@sendgrid/mail';
 
sgMail.setApiKey(process.env.SENDGRID_API_KEY!);
 
interface EmailJobData {
  userId: string;
  email: string;
  name: string;
}
 
export const emailWorker = new Worker<EmailJobData>(
  'email',
  async (job: Job<EmailJobData>) => {
    const { email, name } = job.data;
 
    await sgMail.send({
      to: email,
      from: 'noreply@example.com',
      subject: '환영합니다!',
      text: `안녕하세요, ${name}님. 가입을 환영합니다.`,
    });
 
    return { sentAt: new Date().toISOString() };
  },
  {
    connection,
    concurrency: 10, // I/O 바운드 잡은 vCPU × 2 정도
    metrics: {
      maxDataPoints: MetricsTime.ONE_WEEK, // exportPrometheusMetrics() 시계열 데이터 활성화
    },
  }
);
 
emailWorker.on('completed', (job) => {
  console.log(`이메일 발송 완료: jobId=${job.id}`);
});
 
emailWorker.on('failed', (job, err) => {
  console.error(`이메일 발송 실패: jobId=${job?.id}, error=${err.message}`);
});

// src/queue/dlq.ts
import { Queue, QueueEvents } from 'bullmq';
import { connection } from './config';
 
const SOURCE_QUEUE_NAME = 'email';
 
// 모듈 스코프에서 한 번만 생성해 재사용
// 매 이벤트마다 재생성하면 고부하 시 Redis 연결이 폭증합니다
const sourceQueue = new Queue(SOURCE_QUEUE_NAME, { connection });
const dlqQueue = new Queue(`${SOURCE_QUEUE_NAME}-dlq`, { connection });
const queueEvents = new QueueEvents(SOURCE_QUEUE_NAME, { connection });
 
queueEvents.on('failed', async ({ jobId, failedReason }) => {
  const job = await sourceQueue.getJob(jobId);
 
  if (!job) return;
 
  // 재시도가 아직 남아있는 경우는 DLQ로 이동하지 않음
  const maxAttempts = job.opts.attempts ?? 1;
  if (job.attemptsMade < maxAttempts) return;
 
  // 모든 재시도 소진 후 DLQ로 이동
  await dlqQueue.add(
    'dead-letter',
    {
      originalJobId: job.id,
      originalQueue: SOURCE_QUEUE_NAME,
      jobName: job.name,
      data: job.data,
      failedReason,
      failedAt: new Date().toISOString(),
      attemptsMade: job.attemptsMade,
    },
    {
      removeOnComplete: false, // DLQ 잡은 수동 검토 전까지 보관
      removeOnFail: false,
    }
  );
 
  console.warn(`DLQ로 이동: originalJobId=${job.id}, reason=${failedReason}`);
});

// src/queue/dlq-reprocess.ts
import { Queue } from 'bullmq';
import { connection } from './config';
 
async function reprocessDLQ(limit = 10) {
  const dlqQueue = new Queue('email-dlq', { connection });
  const emailQueue = new Queue('email', { connection });
 
  const waitingJobs = await dlqQueue.getJobs(['waiting'], 0, limit);
 
  for (const dlqJob of waitingJobs) {
    const { data, jobName } = dlqJob.data;
 
    await emailQueue.add(jobName, data, {
      attempts: 5,
      backoff: { type: 'exponential', delay: 3000, jitter: 0.5 },
    });
 
    await dlqJob.remove();
    console.log(`재투입 완료: originalJobId=${dlqJob.data.originalJobId}`);
  }
 
  await dlqQueue.close();
  await emailQueue.close();
}
 
reprocessDLQ();

// src/queue/image.worker.ts
import { Worker } from 'bullmq';
import path from 'path';
import { connection } from './config';
 
export const imageWorker = new Worker(
  'image-processing',
  path.resolve(__dirname, './processors/image.processor.js'), // 컴파일된 JS 파일 참조
  {
    connection,
    concurrency: 2, // CPU 바운드 잡은 1~2로 설정 권장
  }
);

// src/queue/processors/image.processor.ts
import { SandboxedJob } from 'bullmq';
import sharp from 'sharp';
 
// 자식 프로세스 실행 방식이므로 CommonJS module.exports가 요구됩니다
module.exports = async (job: SandboxedJob) => {
  const { inputPath, outputPath, width, height } = job.data;
 
  await sharp(inputPath)
    .resize(width, height, { fit: 'cover' })
    .jpeg({ quality: 85 })
    .toFile(outputPath);
 
  await job.updateProgress(100);
  return { outputPath };
};

// src/metrics/prometheus.ts
import express from 'express';
import { Queue } from 'bullmq';
import { Registry, Gauge, collectDefaultMetrics } from 'prom-client';
import { connection } from '../queue/config';
 
const app = express();
const register = new Registry();
collectDefaultMetrics({ register });
 
const emailQueue = new Queue('email', { connection });
const imageQueue = new Queue('image-processing', { connection });
const queues = [emailQueue, imageQueue];
 
const queueDepthGauge = new Gauge({
  name: 'bullmq_queue_depth',
  help: '큐에 대기 중인 잡 수',
  labelNames: ['queue', 'status'],
  registers: [register],
});
 
app.get('/metrics', async (_req, res) => {
  for (const queue of queues) {
    const counts = await queue.getJobCounts(
      'waiting', 'active', 'completed', 'failed', 'delayed'
    );
 
    for (const [status, count] of Object.entries(counts)) {
      queueDepthGauge.set({ queue: queue.name, status }, count);
    }
  }
 
  const bullMetrics = await Promise.all(
    queues.map((q) => q.exportPrometheusMetrics())
  );
 
  const customMetrics = await register.metrics();
 
  res.set('Content-Type', 'text/plain');
  res.send([customMetrics, ...bullMetrics].join('\n'));
});
 
app.listen(9100, () => {
  console.log('Prometheus 메트릭 서버 시작: http://localhost:9100/metrics');
});

// src/queue/graceful-shutdown.ts
import { emailWorker } from './email.worker';
import { imageWorker } from './image.worker';
 
const workers = [emailWorker, imageWorker];
 
async function shutdown() {
  console.log('종료 신호 수신, 진행 중인 잡 완료 후 종료합니다...');
 
  // close()는 현재 처리 중인 잡이 완료될 때까지 대기 후 Worker를 종료합니다
  await Promise.all(workers.map((w) => w.close()));
 
  console.log('Worker 종료 완료');
  process.exit(0);
}
 
process.on('SIGTERM', shutdown);
process.on('SIGINT', shutdown);