BullMQ로 구축하는 프로덕션 작업 큐 — 우선순위, 재시도 백오프, Dead Letter Queue를 TypeScript로 구현하기

pnpm add bullmq

// config/redis.ts
import { ConnectionOptions } from 'bullmq'
 
export const connection: ConnectionOptions = {
  host: process.env.REDIS_HOST ?? 'localhost',
  port: Number(process.env.REDIS_PORT ?? 6379),
}

// queues/notification.queue.ts
import { Queue } from 'bullmq'
import { connection } from '../config/redis'
 
export interface NotificationJobData {
  userId: string
  type: 'urgent' | 'normal' | 'batch'
  message: string
}
 
export const notificationQueue = new Queue<NotificationJobData>('notifications', {
  connection,
  defaultJobOptions: {
    removeOnComplete: { count: 1000, age: 3600 },
    removeOnFail: { count: 5000 },
  },
})
 
// 우선순위 1: 즉시 처리해야 하는 보안 경고
await notificationQueue.add(
  'urgent-alert',
  { userId: 'u-123', type: 'urgent', message: '계정 보안 경고' },
  { priority: 1 }
)
 
// 우선순위 10: 일반 사용자 알림
await notificationQueue.add(
  'normal-notification',
  { userId: 'u-456', type: 'normal', message: '새 댓글이 달렸습니다' },
  { priority: 10 }
)
 
// 우선순위 100: 배치 마케팅 메일 (가장 낮음)
await notificationQueue.add(
  'batch-email',
  { userId: 'u-789', type: 'batch', message: '이번 주 뉴스레터' },
  { priority: 100 }
)

const job = await notificationQueue.getJob(jobId)
if (job) {
  await job.changePriority({ priority: 1 })
}

await paymentQueue.add(
  'process-payment',
  {
    orderId: 'order-001',
    amount: 15000,
    currency: 'KRW',
    idempotencyKey: crypto.randomUUID(),
  },
  {
    attempts: 5,
    backoff: {
      type: 'exponential',
      delay: 1000,  // 1s → 2s → 4s → 8s → 16s
    },
  }
)

const worker = new Worker('api-calls', processor, {
  connection,
  settings: {
    backoffStrategy: (attemptsMade, type, err, job) => {
      // TypeScript strict 모드에서 err는 unknown — 타입 가드 필요
      if (err instanceof Error && err.message.includes('Rate limit')) {
        return 60_000 * attemptsMade  // Rate limit이면 분 단위로 증가
      }
      return 2 ** attemptsMade * 1000
    },
  },
})

// 핵심 패턴 — 전체 구현은 아래 "실전 구현" 섹션 참고
const dlqQueue = new Queue('payments-dlq', { connection })
 
paymentWorker.on('failed', async (job, err) => {
  if (!job) return  // BullMQ는 failed 이벤트에서 job을 undefined로 넘길 수 있음
 
  const maxAttempts = job.opts.attempts ?? 1
  if (job.attemptsMade >= maxAttempts) {
    await dlqQueue.add('failed-payment', {
      originalJobId: job.id ?? 'unknown',
      originalData: job.data,
      errorMessage: err instanceof Error ? err.message : String(err),
      failedAt: new Date().toISOString(),
    })
  }
})

// queues/payment.queue.ts
import { Queue } from 'bullmq'
import { connection } from '../config/redis'
 
export interface PaymentJobData {
  orderId: string
  amount: number
  currency: string
  idempotencyKey: string
}
 
export const paymentQueue = new Queue<PaymentJobData>('payments', {
  connection,
  defaultJobOptions: {
    attempts: 5,
    backoff: { type: 'exponential', delay: 1000 },
    removeOnComplete: { count: 500, age: 3600 },
    removeOnFail: { count: 2000 },
  },
})

// queues/payment-dlq.queue.ts
import { Queue } from 'bullmq'
import { connection } from '../config/redis'
import type { PaymentJobData } from './payment.queue'
 
export interface DLQJobData {
  originalJobId: string
  originalJobName: string
  originalData: PaymentJobData
  errorMessage: string
  attemptsMade: number
  failedAt: string
}
 
export const paymentDLQQueue = new Queue<DLQJobData>('payments-dlq', { connection })

// workers/payment.worker.ts
import { Worker } from 'bullmq'
import { connection } from '../config/redis'
import type { PaymentJobData } from '../queues/payment.queue'
import { paymentDLQQueue } from '../queues/payment-dlq.queue'
 
const paymentWorker = new Worker<PaymentJobData>(
  'payments',
  async (job) => {
    await job.updateProgress(10)
 
    // callPaymentGateway는 실제 결제 게이트웨이 클라이언트 — 구현체 생략
    const result = await callPaymentGateway({
      orderId: job.data.orderId,
      amount: job.data.amount,
      idempotencyKey: job.data.idempotencyKey,
    })
 
    await job.updateProgress(100)
    return result
  },
  {
    connection,
    concurrency: 5,
    limiter: {
      max: 10,       // 초당 최대 10건 — 결제 게이트웨이 Rate Limit 대응
      duration: 1000,
    },
  }
)
 
paymentWorker.on('failed', async (job, err) => {
  if (!job) return
 
  const maxAttempts = job.opts.attempts ?? 1
 
  if (job.attemptsMade >= maxAttempts) {
    console.error(`[DLQ] 잡 ${job.id} 최종 실패 → DLQ 이송`)
 
    await paymentDLQQueue.add(
      'failed-payment',
      {
        originalJobId: job.id ?? 'unknown',
        originalJobName: job.name,
        originalData: job.data,
        errorMessage: err instanceof Error ? err.message : String(err),
        attemptsMade: job.attemptsMade,
        failedAt: new Date().toISOString(),
      },
      {
        removeOnComplete: { age: 7 * 24 * 3600 },
        // DLQ에 워커를 추가해 자동 재처리할 경우, 그 처리가 실패해도 기록이 자동 삭제되지 않도록 함
        removeOnFail: false,
      }
    )
  }
})
 
paymentWorker.on('completed', (job) => {
  console.log(`[${job.id}] 결제 완료: ${job.data.orderId}`)
})

// scripts/reprocess-dlq.ts
import { Queue } from 'bullmq'
import { connection } from '../config/redis'
import type { DLQJobData } from '../queues/payment-dlq.queue'
 
const dlqQueue = new Queue<DLQJobData>('payments-dlq', { connection })
const paymentQueue = new Queue('payments', { connection })
 
async function listDLQJobs() {
  // DLQ 워커가 없으므로 잡은 'waiting' 상태로 쌓임
  const jobs = await dlqQueue.getJobs(['waiting', 'delayed'])
  return jobs.map((job) => ({
    id: job.id,
    orderId: job.data.originalData.orderId,
    error: job.data.errorMessage,
    attempts: job.data.attemptsMade,
    failedAt: job.data.failedAt,
  }))
}
 
async function reprocessDLQJob(dlqJobId: string) {
  const dlqJob = await dlqQueue.getJob(dlqJobId)
  if (!dlqJob) throw new Error(`DLQ 잡 ${dlqJobId}를 찾을 수 없습니다`)
 
  const newJob = await paymentQueue.add(
    dlqJob.data.originalJobName,
    dlqJob.data.originalData,
    {
      attempts: 3,
      backoff: { type: 'exponential', delay: 2000 },
      priority: 5,  // 재처리는 일반 잡보다 높은 우선순위
    }
  )
 
  await dlqJob.remove()
 
  console.log(`재처리 완료: DLQ ${dlqJobId} → 새 잡 ${newJob.id}`)
  return newJob.id
}

# redis.conf
maxmemory-policy noeviction

// count만 지정: 트래픽이 없는 기간에 몇 달 된 잡이 남아있을 수 있음
// age만 지정: 트래픽 폭증 시 메모리가 급격히 증가할 수 있음
// 둘 다 지정:
removeOnComplete: { count: 1000, age: 3600 }
removeOnFail: { count: 5000 }

// Prisma 기준 예시
async function paymentProcessor(job: Job<PaymentJobData>) {
  const existing = await prisma.payment.findUnique({
    where: { idempotencyKey: job.data.idempotencyKey },
  })
 
  if (existing) {
    return existing  // 이미 처리된 결제: 이전 결과 반환
  }
 
  return await prisma.payment.create({ data: { ...job.data } })
}

// notifications/notifications.module.ts
import { BullModule } from '@nestjs/bullmq'
import { Module } from '@nestjs/common'
import { NotificationProcessor } from './notification.processor'
import { NotificationService } from './notification.service'
 
@Module({
  imports: [
    BullModule.registerQueue({
      name: 'notifications',
      defaultJobOptions: {
        attempts: 3,
        backoff: { type: 'exponential', delay: 1000 },
        removeOnComplete: { count: 1000, age: 3600 },
        removeOnFail: { count: 5000 },
      },
    }),
  ],
  providers: [NotificationService, NotificationProcessor],
  exports: [NotificationService],
})
export class NotificationsModule {}

// notifications/notification.processor.ts
import { Processor, WorkerHost, OnWorkerEvent } from '@nestjs/bullmq'
import { Job } from 'bullmq'
import type { NotificationJobData } from './notification.types'
 
@Processor('notifications', { concurrency: 10 })
export class NotificationProcessor extends WorkerHost {
  async process(job: Job<NotificationJobData>): Promise<void> {
    switch (job.name) {
      case 'send-email':
        // sendEmail, sendPush는 실제 알림 서비스 클라이언트 — 구현체 생략
        await this.sendEmail(job.data)
        break
      case 'send-push':
        await this.sendPush(job.data)
        break
      default:
        throw new Error(`알 수 없는 잡 타입: ${job.name}`)
    }
  }
 
  @OnWorkerEvent('failed')
  onFailed(job: Job<NotificationJobData> | undefined, err: Error) {
    if (!job) return  // failed 이벤트에서 job이 undefined일 수 있음
    console.error(`알림 잡 실패 [${job.id}]: ${err.message}`)
  }
}

// app.ts (Express 기준)
import { createBullBoard } from '@bull-board/api'
import { BullMQAdapter } from '@bull-board/api/bullMQAdapter'
import { ExpressAdapter } from '@bull-board/express'
import { notificationQueue } from './queues/notification.queue'
import { paymentQueue } from './queues/payment.queue'
import { paymentDLQQueue } from './queues/payment-dlq.queue'
 
const serverAdapter = new ExpressAdapter()
serverAdapter.setBasePath('/admin/queues')
 
createBullBoard({
  queues: [
    new BullMQAdapter(notificationQueue),
    new BullMQAdapter(paymentQueue),
    new BullMQAdapter(paymentDLQQueue),
  ],
  serverAdapter,
})
 
app.use('/admin/queues', serverAdapter.getRouter())