BullMQ 실전 튜토리얼 — Node.js 백그라운드 작업 큐 구현, 재시도·스케줄링·모니터링까지

docker-compose.yml — 레디스 실행
version: '3.8'
services:
  redis:
    image: redis:7-alpine
    ports:
      - '6379:6379'
    volumes:
      - redis-data:/data
    command: redis-server --appendonly yes

volumes:
  redis-data:

BullMQ 패키지 설치
# npm
npm install bullmq

# pnpm
pnpm add bullmq

# BullMQ는 타입 정의를 내장하고 있어 @types 별도 설치 불필요
# ioredis는 BullMQ 내부 의존성이지만, 레디스를 직접 다룰 때 설치
npm install ioredis

src/queues/email.queue.ts — 큐 정의
import { Queue } from 'bullmq'

export interface EmailJobData {
  to: string
  subject: string
  body: string
  template?: string
}

const connection = {
  host: process.env.REDIS_HOST || 'localhost',
  port: Number(process.env.REDIS_PORT) || 6379,
}

export const emailQueue = new Queue<EmailJobData>('email', {
  connection,
  defaultJobOptions: {
    attempts: 3,           // 최대 재시도 횟수
    backoff: {
      type: 'exponential',
      delay: 2000,         // 첫 재시도 2초 후, 이후 2배씩 증가
    },
    removeOnComplete: 100, // 완료된 잡은 최근 100개만 레디스에 보관
    removeOnFail: 500,     // 실패 잡은 최근 500개 보관
  },
})

// API 핸들러에서 호출
export async function enqueueEmail(data: EmailJobData) {
  const job = await emailQueue.add('send-email', data)
  console.log(`이메일 잡 추가 완료 — job id: ${job.id}`)
  return job
}

src/workers/email.worker.ts — 워커 정의
import { Worker, Job } from 'bullmq'
import { EmailJobData } from '../queues/email.queue'
import { sendEmail } from '../lib/mailer' // 실제 메일 발송 함수

const connection = {
  host: process.env.REDIS_HOST || 'localhost',
  port: Number(process.env.REDIS_PORT) || 6379,
}

const emailWorker = new Worker<EmailJobData>(
  'email',
  async (job: Job<EmailJobData>) => {
    const { to, subject, body, template } = job.data

    // 진행 상태 업데이트 — Bull Board에서 확인 가능
    await job.updateProgress(10)

    await sendEmail({ to, subject, body, template })

    await job.updateProgress(100)
    console.log(`이메일 발송 완료 — to: ${to}`)
  },
  {
    connection,
    concurrency: 5, // 워커 하나가 동시에 처리할 잡 수
  }
)

emailWorker.on('completed', (job) => {
  console.log(`잡 완료 — id: ${job.id}`)
})

emailWorker.on('failed', (job, err) => {
  console.error(`잡 실패 — id: ${job?.id}, 오류: ${err.message}`)
})

export default emailWorker

재시도 전략 설정 예시
// 큐 기본값 설정 — 모든 잡에 적용
const emailQueue = new Queue('email', {
  connection,
  defaultJobOptions: {
    attempts: 5,      // 최초 실행 포함 총 5회
    backoff: {
      type: 'exponential',
      delay: 1000,    // 1초 → 2초 → 4초 → 8초 순서로 증가
    },
  },
})

// 개별 잡에 별도 옵션 지정 — 큐 기본값을 덮어씀
await emailQueue.add('send-email', data, {
  attempts: 3,
  backoff: { type: 'fixed', delay: 5000 },
})

// 커스텀 백오프 — 워커 설정에서 지정
const worker = new Worker('email', processor, {
  connection,
  settings: {
    backoffStrategy: (attemptsMade, type, err) => {
      // 429 Too Many Requests는 30초 후 재시도
      if (err?.message?.includes('429')) return 30000
      // 그 외는 지수 증가
      return Math.pow(2, attemptsMade) * 1000
    },
  },
})

지연 실행과 cron 스케줄링
import { Queue } from 'bullmq'

const queue = new Queue('notifications', { connection })

// 지연 실행 — 1시간 후 처리 (밀리초 단위)
await queue.add(
  'reminder-email',
  { userId: '12345', type: 'signup-followup' },
  { delay: 60 * 60 * 1000 }
)

// cron 반복 — 매일 새벽 2시 KST
await queue.add(
  'daily-report',
  { reportType: 'daily-summary' },
  {
    repeat: {
      pattern: '0 2 * * *',
      tz: 'Asia/Seoul',
    },
  }
)

// 3시간마다 반복 (인터벌 방식)
await queue.add(
  'cache-warmup',
  { cacheKey: 'popular-posts' },
  {
    repeat: { every: 3 * 60 * 60 * 1000 },
  }
)

// 등록된 반복 잡 목록 확인
const repeatableJobs = await queue.getRepeatableJobs()

// 특정 반복 잡 제거
await queue.removeRepeatableByKey(repeatableJobs[0].key)

Bull Board Express 연동
import express from 'express'
import { createBullBoard } from '@bull-board/api'
import { BullMQAdapter } from '@bull-board/api/bullMQAdapter'
import { ExpressAdapter } from '@bull-board/express'
import { emailQueue } from './queues/email.queue'
import { notificationQueue } from './queues/notification.queue'

// 패키지 설치
// npm install @bull-board/api @bull-board/express

const app = express()

const serverAdapter = new ExpressAdapter()
serverAdapter.setBasePath('/admin/queues')

createBullBoard({
  queues: [
    new BullMQAdapter(emailQueue),
    new BullMQAdapter(notificationQueue),
  ],
  serverAdapter,
})

// 대시보드 라우트 마운트
app.use('/admin/queues', serverAdapter.getRouter())

app.listen(3000, () => {
  console.log('대시보드: http://localhost:3000/admin/queues')
})

Graceful shutdown 처리
import { Worker } from 'bullmq'

const worker = new Worker('email', processor, { connection })

// SIGTERM 처리 — 쿠버네티스, PM2, Docker 배포 시 중요
process.on('SIGTERM', async () => {
  console.log('SIGTERM 수신 — 워커 종료 중...')

  // 현재 처리 중인 잡이 완료될 때까지 대기 후 정상 종료
  await worker.close()

  console.log('워커 종료 완료')
  process.exit(0)
})

process.on('SIGINT', async () => {
  await worker.close()
  process.exit(0)
})