BullMQ 5 + Redis로 Node.js 백그라운드 작업 큐 운영하기

pnpm add bullmq ioredis

// lib/redis.ts
import { Redis } from 'ioredis';
 
export const redisConnection = new Redis({
  host: process.env.REDIS_HOST ?? 'localhost',
  port: Number(process.env.REDIS_PORT ?? 6379),
  maxRetriesPerRequest: null,
});

// queues/email.queue.ts
import { Queue } from 'bullmq';
import { redisConnection } from '../lib/redis';
 
export const emailQueue = new Queue('email', {
  connection: redisConnection,
  defaultJobOptions: {
    removeOnComplete: { count: 1_000, age: 24 * 3600 },
    removeOnFail:     { count: 5_000, age: 7 * 24 * 3600 },
  },
});

pnpm add @bull-board/api @bull-board/express

// app.ts
import { createBullBoard } from '@bull-board/api';
import { BullMQAdapter } from '@bull-board/api/bullMQAdapter';
import { ExpressAdapter } from '@bull-board/express';
import basicAuth from 'express-basic-auth';
import { emailQueue } from './queues/email.queue';
import { paymentQueue } from './queues/payment.queue';
 
const serverAdapter = new ExpressAdapter();
serverAdapter.setBasePath('/admin/queues');
 
createBullBoard({
  queues: [
    new BullMQAdapter(emailQueue),
    new BullMQAdapter(paymentQueue),
  ],
  serverAdapter,
});
 
// 프로덕션에서는 인증 미들웨어를 반드시 앞에 붙이세요.
app.use(
  '/admin/queues',
  basicAuth({
    users: { admin: process.env.BULL_BOARD_PASSWORD ?? '' },
    challenge: true,
  }),
  serverAdapter.getRouter(),
);

// VIP 사용자 트랜잭션 메일 — 즉시 처리
await emailQueue.add('transactional', { to: vipUser.email }, { priority: 1 });
 
// 일반 알림 메일
await emailQueue.add('notification', { to: user.email }, { priority: 5 });
 
// 마케팅 뉴스레터 — 여유 있을 때
await emailQueue.add('newsletter', { to: user.email }, { priority: 10 });

// 회원가입 5분 후 웰컴 이메일
await emailQueue.add(
  'welcome',
  { to: newUser.email, name: newUser.name },
  { delay: 5 * 60 * 1_000 },
);
 
// 결제 1시간 후 영수증 발송
await emailQueue.add(
  'receipt',
  { orderId, to: user.email },
  { delay: 60 * 60 * 1_000 },
);
 
// 매주 월요일 오전 9시 리포트 — 한국 시각 기준
await emailQueue.add(
  'weekly-report',
  { type: 'summary' },
  {
    repeat: {
      pattern: '0 9 * * MON',
      tz: 'Asia/Seoul',
    },
  },
);

await paymentQueue.add('confirm-payment', { orderId }, {
  attempts: 5,
  backoff: {
    type: 'exponential',
    delay: 1_000,
    // 1초 → 2초 → 4초 → 8초 → 16초
  },
});

import { UnrecoverableError } from 'bullmq';
 
async function processor(job: Job) {
  try {
    return await riskyApiCall(job.data);
  } catch (error) {
    if (error instanceof ValidationError) {
      // 재시도해도 달라지지 않는 오류 — 즉시 최종 실패 처리
      throw new UnrecoverableError(`검증 오류: ${error.message}`);
    }
    throw error; // 일시적 오류 — 재시도 트리거
  }
}

import { Worker, Job } from 'bullmq';
 
const worker = new Worker('payment', processor, {
  settings: {
    backoffStrategy: (
      attemptsMade: number,
      _type: string,
      _err: Error,
      _job: Job,
    ): number => {
      const base = Math.min(attemptsMade * 2_000, 30_000);
      const jitter = Math.random() * 1_000;
      return base + jitter;
    },
  },
  connection: redisConnection,
});

// queues/payment.queue.ts
import { Queue } from 'bullmq';
import { redisConnection } from '../lib/redis';
 
export interface PaymentJobData {
  orderId: string;
  amount: number;
  userId: string;
}
 
export const paymentQueue = new Queue<PaymentJobData>('payment', {
  connection: redisConnection,
  defaultJobOptions: {
    attempts: 5,
    backoff: { type: 'exponential', delay: 2_000 },
    removeOnComplete: { count: 500 },
    removeOnFail: { count: 2_000, age: 7 * 24 * 3600 },
  },
});
 
export const dlqQueue = new Queue('payment-dlq', {
  connection: redisConnection,
});

// workers/payment.worker.ts
import { Worker, Job, UnrecoverableError } from 'bullmq';
import { PaymentJobData, paymentQueue, dlqQueue } from '../queues/payment.queue';
import { pgApiClient } from '../lib/pg-client';
import { notifyOncall } from '../lib/notifications';
import { redisConnection } from '../lib/redis';
 
async function processPayment(job: Job<PaymentJobData>) {
  const { orderId, amount } = job.data;
 
  await job.updateProgress(10);
 
  try {
    const result = await pgApiClient.confirmPayment({ orderId, amount });
    await job.updateProgress(100);
    return result;
  } catch (error) {
    if (error instanceof InvalidOrderError) {
      // 재시도해도 달라지지 않는 오류 — 즉시 DLQ로
      throw new UnrecoverableError(`주문 오류: ${(error as Error).message}`);
    }
    throw error;
  }
}
 
const worker = new Worker<PaymentJobData>('payment', processPayment, {
  connection: redisConnection,
  concurrency: 5,
  stalledInterval: 30_000,  // 30초마다 stall 감지
  maxStalledCount: 2,       // 2번 stall 이후 failed 처리
});

// 결제 확인 작업 등록 — orderId로 중복 방지
await paymentQueue.add(
  'confirm-payment',
  { orderId, amount, userId },
  { jobId: orderId },
);

worker.on('failed', async (job, error) => {
  if (!job) return;
 
  const maxAttempts = job.opts.attempts ?? 1;
  const isExhausted =
    job.attemptsMade >= maxAttempts || error instanceof UnrecoverableError;
 
  if (isExhausted) {
    try {
      await dlqQueue.add('failed-payment', {
        originalJobId: job.id,
        originalData: job.data,
        error: error.message,
        failedAt: new Date().toISOString(),
      });
      await notifyOncall(`결제 작업 최종 실패: orderId=${job.data.orderId}`);
    } catch (dlqError) {
      console.error('DLQ 등록 실패:', dlqError);
    }
  }
});
 
// 그레이스풀 셧다운 — 배포 중 작업 유실 방지
process.on('SIGTERM', async () => {
  await worker.close();
  await paymentQueue.close();
  process.exit(0);
});

const twilioWorker = new Worker('sms', async (job: Job) => {
  const { to, body } = job.data;
  await twilioClient.messages.create({
    to,
    body,
    from: process.env.TWILIO_FROM,
  });
}, {
  connection: redisConnection,
  concurrency: 10,
  limiter: {
    max: 100,        // 최대 100건
    duration: 1_000, // 1초 기준
  },
});

// FlowProducer로 파이프라인 등록
import { FlowProducer } from 'bullmq';
 
const flow = new FlowProducer({ connection: redisConnection });
 
await flow.add({
  name: 'store-result',      // 부모 — 마지막에 실행
  queueName: 'pipeline',
  data: { step: 'store' },
  children: [
    {
      name: 'transform-data',
      queueName: 'pipeline',
      data: { step: 'transform' },
      children: [
        {
          name: 'fetch-data', // 최하위 자식 — 가장 먼저 실행
          queueName: 'pipeline',
          data: { step: 'fetch', source: 'api' },
        },
      ],
    },
  ],
});

// workers/pipeline.worker.ts
import { Worker, Job } from 'bullmq';
import { redisConnection } from '../lib/redis';
 
const pipelineWorker = new Worker('pipeline', async (job: Job) => {
  switch (job.name) {
    case 'fetch-data': {
      const raw = await fetchFromApi(job.data.source);
      return raw; // 부모가 getChildrenValues()로 접근
    }
    case 'transform-data': {
      const childValues = await job.getChildrenValues();
      const rawData = Object.values(childValues)[0];
      return transform(rawData);
    }
    case 'store-result': {
      const childValues = await job.getChildrenValues();
      const transformed = Object.values(childValues)[0];
      await saveToDatabase(transformed);
      break;
    }
  }
}, { connection: redisConnection });