BullMQ 패턴
빠른 가이드: BullMQ(v5.x)를 사용하여 Redis 위에서 백그라운드 작업 처리, 작업 스케줄링, 워크플로우 오케스트레이션을 수행하세요. 핵심 클래스:
Queue(작업 추가),Worker(작업 처리),QueueEvents(전역 이벤트 리스너),FlowProducer(부모-자식 작업 트리). 모든 생성자에 항상connection객체를 전달하세요(v5에서 필수). Worker를 위한 ioredis 연결에는maxRetriesPerRequest: null을 설정하세요. 반복 가능/크론 작업에는upsertJobScheduler를 사용하세요(더 이상 사용되지 않는 repeatable API를 대체). QueueScheduler는 v4에서 제거되었으며, 그 역할은 이제 Worker가 자동으로 처리합니다.
<critical_requirements>
중요: 이 스킬 사용 전 필수 사항
모든 코드는 CLAUDE.md의 프로젝트 규칙을 따라야 합니다 (kebab-case, 명명된 내보내기, 임포트 순서,
import type, 명명된 상수)
(모든 Queue, Worker, QueueEvents, FlowProducer 생성자에 connection 객체를 반드시 전달해야 합니다. BullMQ v5는 connection이 없으면 오류를 발생시킵니다.)
(Worker가 사용하는 ioredis 연결에는 반드시 maxRetriesPerRequest: null을 설정해야 합니다. BullMQ는 무한 재시도를 요구하며, 이 설정이 없으면 오류를 발생시킵니다.)
(SIGTERM/SIGINT 시 정상 종료를 위해 반드시 await worker.close()를 호출해야 합니다. 그렇지 않으면 진행 중인 작업이 중단(stalled) 상태가 됩니다.)
(반복 가능/크론 작업에는 반드시 upsertJobScheduler를 사용하세요. queue.add의 repeat 옵션은 v5.16.0부터 더 이상 사용되지 않습니다.)
</critical_requirements>
예제
- 핵심 패턴 -- Queue 설정, Worker 처리, 작업 옵션, 연결 팩토리, 정상 종료, 타입이 있는 작업
- 고급 패턴 -- FlowProducer, 속도 제한, 작업 스케줄링, QueueEvents, 동시성, 샌드박스 처리기
추가 자료:
- reference.md -- 결정 프레임워크, 작업 옵션 참조, 안티패턴, 프로덕션 체크리스트
자동 감지: BullMQ, bullmq, Queue, Worker, QueueEvents, FlowProducer, job queue, background job, worker process, job scheduler, upsertJobScheduler, rate limiter, job priority, job delay, sandboxed processor, repeatable job, cron job, flow producer, parent child jobs
사용 시점:
- 백그라운드 처리 (이메일 전송, 이미지 처리, PDF 생성)
- 예약/크론 작업 (야간 보고서, 정기적인 정리)
- 부모-자식 작업 종속성이 있는 워크플로우 오케스트레이션 (FlowProducer)
- 속도 제한이 있는 API 사용 (아웃바운드 요청 조절)
- 우선순위 기반 작업 처리 (대량 작업보다 긴급 작업 우선)
- 여러 Worker 또는 시스템에 CPU 집약적 작업 분산
다루는 핵심 패턴:
- 타입이 있는 작업 데이터와 반환 값을 사용한 Queue 및 Worker 설정
- Worker용
maxRetriesPerRequest: null을 포함한 연결 팩토리 - 작업 옵션: 지연, 우선순위, 재시도, 백오프, removeOnComplete/Fail
- 프로세스 신호 시
worker.close()를 사용한 정상 종료 - 종속성 추적을 위한 부모-자식 작업 트리를 지원하는 FlowProducer
- 반복/크론 작업을 위한 작업 스케줄러(
upsertJobScheduler) - 속도 제한 (전역 제한기 및 수동
Worker.RateLimitError) - 동시성 제어 (로컬 Worker별 및 전역)
- 모든 Worker에 대한 전역 이벤트 모니터링을 위한 QueueEvents
- CPU 집약적 작업을 위한 샌드박스 처리기
사용하지 말아야 할 때:
- 지속성이 필요 없는 단순한 인-프로세스 타이머 또는
setTimeout - 지속성이 필요 없는 실시간 pub/sub 메시징 (pub/sub 솔루션 사용)
- 요청-응답 주기 내에서 동기적으로 처리되어야 하는 데이터
- 지속성, 재시도 또는 스케줄링이 필요 없는 큐
철학
BullMQ는 Node.js용 Redis 기반 작업 큐로, 최소 한 번 이상 전달 보장(at-least-once delivery)과 함께 안정적인 백그라운드 처리를 제공합니다. 핵심 원칙: 작업 생산과 소비를 분리하여 애플리케이션이 비동기적으로 작업이 수행되는 동안 응답성을 유지하는 것입니다.
핵심 원칙:
- 생산자와 소비자는 분리되어 있습니다. -- 모든 프로세스가 큐에 작업을 추가할 수 있으며, 모든 Worker가 작업을 처리할 수 있습니다. 이를 통해 더 많은 Worker를 추가하여 수평적 확장이 가능합니다.
- 작업은 영구적입니다. -- 작업은 Redis에 저장되므로 프로세스 재시작에도 유지됩니다. 충돌한 Worker의 작업은 다른 Worker(또는 재시작 후 동일한 Worker)가 처리합니다.
- 최소 한 번 이상 전달 보장 -- BullMQ는 모든 작업이 적어도 한 번 처리됨을 보장합니다. 중단(stall) 후 중복 처리가 발생하는 (드문) 경우를 처리하기 위해 멱등성 처리기(idempotent processors)를 사용하세요.
- 재시도를 통한 정상적인 오류 처리 --
attempts및backoff전략을 구성하여 일시적인 오류가 자동으로 해결되도록 합니다. 영구적으로 실패한 작업은 검사를 위해 실패 세트로 이동합니다. - 각 Queue/Worker/QueueEvents는 자체 연결이 필요합니다. -- BullMQ는 내부적으로 연결 상태를 관리합니다. 여러 BullMQ 클래스에서 단일 ioredis 인스턴스를 공유하지 마십시오(프로듀서 역할만 하는 Queue는 제외).
핵심 패턴
패턴 1: 연결 팩토리
BullMQ v5는 모든 생성자에 명시적인 Redis 연결이 필요합니다. Worker는 maxRetriesPerRequest: null이 필요하므로 ioredis가 포기하는 대신 무기한 재시도합니다.
import Redis from "ioredis";
function createBullMQConnection(): Redis {
const url = process.env.REDIS_URL;
if (!url) throw new Error("REDIS_URL is required");
return new Redis(url, { maxRetriesPerRequest: null });
}
export { createBullMQConnection };
왜 좋은가: maxRetriesPerRequest: null은 BullMQ의 요구 사항을 충족하고, 팩토리는 일관된 구성을 보장하며, 환경 변수는 자격 증명을 코드 밖에 유지합니다.
// 잘못된 예 -- maxRetriesPerRequest 누락
const redis = new Redis("redis://localhost:6379");
const worker = new Worker("emails", processor, { connection: redis });
// BullMQ 오류 발생: "maxRetriesPerRequest must be null"
왜 나쁜가: BullMQ는 Worker 연결에 무한 재시도를 요구하며, null 없이 시작 시 오류를 발생시킵니다.
전체 연결 팩토리(프로듀서와 소비자 분리 포함)는 examples/core.md를 참조하세요.
패턴 2: Queue 및 Worker 설정
Queue는 작업을 추가하고, Worker는 작업을 처리합니다. 둘 다 TypeScript 제네릭을 사용하여 타입 안전한 작업 데이터와 반환 값을 지원합니다.
import { Queue, Worker, type Job } from "bullmq";
interface EmailJobData {
to: string;
subject: string;
body: string;
}
const QUEUE_NAME = "emails";
const emailQueue = new Queue<EmailJobData>(QUEUE_NAME, {
connection: createBullMQConnection(),
});
const emailWorker = new Worker<EmailJobData>(
QUEUE_NAME,
async (job: Job<EmailJobData>) => {
await sendEmail(job.data.to, job.data.subject, job.data.body);
},
{ connection: createBullMQConnection() },
);
왜 좋은가: 제네릭은 job.data에 타입 안전성을 강제하고, Queue와 Worker에 대해 별도의 연결을 사용하며, 큐 이름에 명명된 상수를 사용합니다.
전체 설정(이벤트, 오류 처리, 타입이 있는 반환 값 포함)은 examples/core.md를 참조하세요.
패턴 3: 작업 옵션
옵션을 사용하여 지연, 우선순위, 재시도, 백오프 및 자동 제거와 같은 작업 동작을 제어합니다.
const MAX_ATTEMPTS = 5;
const BACKOFF_DELAY_MS = 1000;
const DELAY_MS = 60_000;
const KEEP_COMPLETED_COUNT = 100;
await emailQueue.add(
"send-welcome",
{ to: "[email protected]", subject: "Welcome", body: "..." },
{
delay: DELAY_MS,
priority: 1, // 1 = 가장 높음, 숫자가 높을수록 우선순위가 낮음
attempts: MAX_ATTEMPTS,
backoff: { type: "exponential", delay: BACKOFF_DELAY_MS },
removeOnComplete: { count: KEEP_COMPLETED_COUNT },
removeOnFail: false,
},
);
왜 좋은가: 모든 숫자 값에 명명된 상수를 사용하고, 일시적 오류에 지수 백오프를 사용하며, removeOnComplete with count는 Redis 메모리 무한 증가를 방지하고, removeOnFail: false는 디버깅을 위해 실패한 작업을 보존합니다.
defaultJobOptions, addBulk 및 LIFO 패턴은 examples/core.md를 참조하세요.
패턴 4: 정상 종료
프로세스 신호 시 worker.close()를 호출하여 종료 전에 진행 중인 작업을 완료합니다. 이렇게 하지 않으면 작업이 중단(stalled)되어 다른 Worker에 의해 재처리됩니다.
async function shutdown(workers: Worker[]): Promise<void> {
await Promise.all(workers.map((w) => w.close()));
process.exit(0);
}
process.on("SIGTERM", () => shutdown([emailWorker]));
process.on("SIGINT", () => shutdown([emailWorker]));
왜 좋은가: close()는 새 작업 수락을 중지하고 진행 중인 작업이 완료될 때까지 기다리며, Promise.all은 여러 Worker를 처리하고, 시그널 핸들러는 SIGTERM 및 SIGINT를 모두 처리합니다.
주의사항: worker.close()에는 내장 타임아웃이 없습니다. 프로세서가 중단되면 종료도 중단됩니다. 필요한 경우 자체 타임아웃으로 래핑하세요.
타임아웃 패턴이 있는 종료는 examples/core.md를 참조하세요.
패턴 5: FlowProducer (부모-자식 작업)
FlowProducer는 부모 작업이 모든 자식이 완료될 때까지 기다리는 작업 트리를 생성합니다. 전체 트리가 원자적으로 추가됩니다.
import { FlowProducer } from "bullmq";
const flowProducer = new FlowProducer({ connection: createBullMQConnection() });
await flowProducer.add({
name: "publish-video",
queueName: "videos",
children: [
{ name: "extract-audio", queueName: "media", data: { videoId: "abc" } },
{
name: "generate-thumbnail",
queueName: "media",
data: { videoId: "abc" },
},
{
name: "transcode",
queueName: "media",
data: { videoId: "abc", format: "mp4" },
},
],
});
왜 좋은가: 부모는 모든 자식이 완료될 때까지 기다리고, 원자적 추가(전부 아니면 전무)이며, 자식은 다른 큐에 있을 수 있고, 부모 프로세서는 job.getChildrenValues()를 통해 자식 결과에 액세스할 수 있습니다.
자식 값 액세스 및 중첩된 흐름은 examples/advanced.md를 참조하세요.
패턴 6: 작업 스케줄러 (반복/크론 작업)
반복 가능한 작업에는 upsertJobScheduler(v5.16.0+)을 사용합니다. 더 이상 사용되지 않는 repeat 옵션을 대체합니다.
const REPORT_INTERVAL_MS = 3_600_000; // 1시간
// 고정 간격
await emailQueue.upsertJobScheduler(
"hourly-digest",
{ every: REPORT_INTERVAL_MS },
{
name: "send-digest",
data: { type: "hourly" },
},
);
// Cron 표현식 -- 매일 오전 3시 15분
await emailQueue.upsertJobScheduler(
"nightly-cleanup",
{ pattern: "0 15 3 * * *" },
{
name: "cleanup",
data: { type: "nightly" },
opts: { removeOnComplete: true },
},
);
왜 좋은가: upsertJobScheduler는 멱등성(idempotent)이며(매 시작 시 호출해도 안전), 템플릿은 각 생성된 작업이 상속하는 작업 이름/데이터/옵션을 정의하며, pattern을 통한 cron 구문을 제공합니다.
주의사항: every 간격은 upsertJobScheduler를 호출한 시간이 아니라 시계에 맞춰 정렬됩니다. 2000ms 간격은 0초, 2초, 4초 등에 실행됩니다.
스케줄러 제거 및 활성 스케줄러 나열은 examples/advanced.md를 참조하세요.
패턴 7: 속도 제한
Worker가 시간 창당 처리하는 작업 수를 제한합니다. 제한기는 동일한 큐의 모든 Worker에 대해 전역적으로 적용됩니다.
const RATE_LIMIT_MAX = 10;
const RATE_LIMIT_DURATION_MS = 1000;
const worker = new Worker("api-calls", processor, {
connection: createBullMQConnection(),
limiter: { max: RATE_LIMIT_MAX, duration: RATE_LIMIT_DURATION_MS },
});
왜 좋은가: 전역 속도 제한(이 구성을 사용하는 10개의 Worker는 여전히 최대 10작업/초만 처리), 제한에 대해 명명된 상수 사용
외부 API 응답에 따른 동적 속도 제한을 위해서는 worker.rateLimit(duration) + throw Worker.RateLimitError()를 사용하세요. examples/advanced.md를 참조하세요.
패턴 8: 전역 모니터링을 위한 QueueEvents
QueueEvents는 Redis Streams(Pub/Sub이 아님)를 사용하므로 재연결 시에도 이벤트가 안정적으로 전달됩니다. 자체 전용 연결이 필요합니다.
import { QueueEvents } from "bullmq";
const queueEvents = new QueueEvents(QUEUE_NAME, {
connection: createBullMQConnection(),
});
queueEvents.on("completed", ({ jobId, returnvalue }) => {
console.log(`작업 ${jobId} 완료 (결과: ${returnvalue})`);
});
queueEvents.on("failed", ({ jobId, failedReason }) => {
console.error(`작업 ${jobId} 실패: ${failedReason}`);
});
왜 좋은가: 단일 리스너로 큐의 모든 Worker를 모니터링하고, Redis Streams를 통한 안정적인 전달, 필요에 따라 별도의 연결을 사용합니다.
진행률 추적 및 특정 작업 완료 대기는 examples/advanced.md를 참조하세요.
성능 최적화
- 동시성 (Concurrency) -- 단일 Worker 인스턴스 내에서 여러 작업을 병렬로 처리하려면 Worker 옵션에
concurrency를 설정하세요:{ concurrency: 10 }. I/O 바운드 작업에만 효과적입니다. CPU 바운드 작업은 이벤트 루프를 차단하고 작업 중단(stalled jobs)을 유발합니다. 대신 샌드박스 처리기를 사용하세요. - 샌드박스 처리기 (Sandboxed processors) -- Worker 생성자에 함수 대신 파일 경로를 전달하여 별도의 프로세스(또는
useWorkerThreads: true를 사용한 작업자 스레드)에서 프로세서를 실행합니다. CPU 집약적 작업이 잠금 갱신을 차단하는 것을 방지합니다. - addBulk --
queue.add()를 반복 호출하는 대신queue.addBulk([...])를 사용하여 단일 Redis 왕복으로 많은 작업을 추가하세요. - removeOnComplete/Fail -- 무제한 Redis 메모리 증가를 방지하려면 자동 제거를 구성하세요. 디버깅을 위해 마지막 N개 작업을 유지하려면
{ count: N }을 사용하세요. - 별도 연결 -- 프로듀서(Queue)는 연결을 공유할 수 있습니다. Worker 및 QueueEvents는 차단 Redis 명령으로 인해 각각 자체 연결이 필요합니다.
<decision_framework>
결정 프레임워크
BullMQ를 사용해야 하는 경우
백그라운드 작업 처리가 필요합니까?
|-- 아니오 -> BullMQ를 사용하지 마십시오
+-- 예 -> 지속성, 재시도 또는 스케줄링이 필요합니까?
|-- 아니오 -> 단순한 인-프로세스 큐 또는 setTimeout으로 충분할 수 있음
+-- 예 -> 부모-자식 작업 종속성이 필요합니까?
|-- 예 -> FlowProducer와 함께 BullMQ 사용
+-- 아니오 -> 속도 제한 또는 우선순위가 필요합니까?
|-- 예 -> 제한기/우선순위 옵션과 함께 BullMQ 사용
+-- 아니오 -> 기본 Queue + Worker로 BullMQ 사용
어떤 작업 패턴을 사용해야 합니까?
어떤 종류의 작업 스케줄링이 필요합니까?
|-- 일회성 지연 작업 -> delay 옵션과 함께 queue.add() 사용
|-- 반복/크론 작업 -> upsertJobScheduler() 사용
|-- 부모-자식 종속성 -> FlowProducer 사용
|-- 속도 제한이 필요한 API 호출 -> Worker limiter 옵션 사용
|-- 우선순위 기반 처리 -> priority 옵션과 함께 queue.add() 사용
</decision_framework>
설정 참조
기본 Queue 및 Worker 설정
| 설정 | 권장 값 | 설명 |
|---|---|---|
connection (모든 BullMQ 클래스) | 전용 ioredis 인스턴스 | Worker의 경우 maxRetriesPerRequest: null 설정 |
concurrency (Worker) | 10 | I/O 바운드 작업에 대한 기본 병렬 처리 |
limiter (Worker) | { max: 10, duration: 1000 } | 작업 속도를 초당 10개로 제한 |
defaultJobOptions (Queue) | { attempts: 3, backoff: { type: 'exponential', delay: 1000 } } | 모든 작업에 대한 기본 재시도 정책 |
작업 옵션
| 옵션 | 유형 | 설명 |
|---|---|---|
delay | number (ms) | 작업 처리를 지연시킵니다 |
priority | number | 1이 가장 높음, 숫자가 높을수록 우선순위가 낮음 |
attempts | number | 실패한 작업의 최대 재시도 횟수 |
backoff | object | 재시도 사이의 지연 전략 (type: "fixed" 또는 "exponential") |
removeOnComplete | object / boolean | 완료된 작업 자동 제거 ({ count: N } 또는 { age: seconds }) |
removeOnFail | object / boolean | 실패한 작업 자동 제거 ({ count: N } 또는 { age: seconds }) |
제거 옵션
| 패턴 | 사용 사례 |
|---|---|
{ count: 100 } | 디버깅을 위해 마지막 100개 완료 작업 유지 |
{ age: 86400 } | 24시간(86400초)이 지난 완료 작업 제거 |
false | 실패한 작업을 무기한 보존 |
true | 완료/실패 즉시 작업 제거 (메모리 최적화) |
설치
npm install bullmq ioredis
BullMQ v5.x에는 Node.js 16+ 및 Redis 6.2+가 필요합니다.
리소스
예제
- 핵심 패턴 -- Queue 설정, Worker 처리, 작업 옵션, 연결 팩토리, 정상 종료, 타입이 있는 작업
- 고급 패턴 -- FlowProducer, 속도 제한, 작업 스케줄링, QueueEvents, 동시성, 샌드박스 처리기
참조
- reference.md -- 결정 프레임워크, 작업 옵션 참조, 안티패턴, 프로덕션 체크리스트
안티패턴 (피해야 할 것)
- 연결 공유 --
Queue,Worker,QueueEvents간에 단일ioredis인스턴스를 공유하지 마십시오. 각각 자체 인스턴스가 필요합니다. - maxRetriesPerRequest 누락 -- Worker에 필요한
maxRetriesPerRequest: null을 생략하면 시작 시 오류가 발생합니다. - 정상 종료 누락 --
worker.close()를 호출하지 않으면 진행 중인 작업이 중단 상태가 됩니다. - 무분별한
removeOnComplete: false-- 완료된 작업을 영원히 유지하면 Redis 메모리가 소진됩니다. 항상{ count: N }을 설정하세요. - 긴 실행 동기 프로세서 -- 동기식 CPU 집약적 작업은 이벤트 루프를 차단하고 잠금 갱신을 방지합니다. 항상 샌드박스 처리기를 사용하세요.
- Worker의 오류 처리되지 않음 -- Worker 프로세서의 포착되지 않은 오류로 인해 작업이 계속 재시도됩니다. 항상 오류를 적절히 처리하세요.
<production_checklist>
프로덕션 체크리스트
- 모든 BullMQ 클래스(Queue, Worker, QueueEvents, FlowProducer)에
connection이 전달됩니까? - 모든 Worker 연결에
maxRetriesPerRequest: null이 설정되었습니까? - 정상 종료가 구현되었습니까? (
worker.close()onSIGTERM/SIGINT) - 완료/실패 시 작업이 제거되도록 설정되었습니까? (
removeOnComplete/removeOnFail) - 재시도 정책이 구성되었습니까? (
attempts+backoff) - 작업 데이터가 JSON 직렬화 가능합니까? (BullMQ는 JSON으로 직렬화함)
- 프로세서가 멱등성(idempotent)입니까? (중복 처리 시나리오 처리)
- Redis 연결이 풀링되었습니까? (
ioredis는 기본적으로 연결을 풀링함) - QueueEvents 연결이 전용입니까? (다른 BullMQ 클래스와 공유하지 않음)
</production_checklist>