sickn33/bullmq-specialist

bullmq-specialist은(는) 무엇을 하나요?

BullMQ 전문가: Redis 기반 작업 큐, 백그라운드 처리, Node.js/TypeScript 애플리케이션에서의 안정적인 비동기 실행.

BullMQ는 Node.js에서 Redis를 기반으로 한 강력한 작업 큐 라이브러리입니다. 이 스킬은 BullMQ를 사용하여 작업 큐를 설정하고, 작업을 추가 및 처리하며, 오류를 처리하고, 작업 실행을 모니터링하는 방법에 대한 모범 사례를 제공합니다.

주요 기능

• 작업 큐 생성 및 구성: BullMQ Queue와 Worker 클래스를 사용하여 작업 큐를 생성합니다.
• 작업 추가: add(), addBulk() 등의 메서드를 사용하여 작업을 큐에 추가합니다.
• 작업 처리: Worker와 Job 객체를 사용하여 작업을 처리하고 진행 상황을 보고합니다.
• 재시도 및 지연: 작업 실패 시 자동 재시도와 지연 실행을 설정합니다.
• 작업 스케줄링: QueueScheduler를 사용하여 반복 작업을 예약합니다.
• 작업 순서화 및 우선순위: 작업의 실행 순서와 우선순위를 제어합니다.
• Redis 연결 관리: IORedis 연결과 BullMQ의 연결 풀을 관리합니다.
• 모니터링: QueueEvents를 사용하여 작업 이벤트를 추적하고 모니터링합니다.

시작하기

다음 명령어로 BullMQ를 설치하세요:

npm install bullmq ioredis
# 또는
yarn add bullmq ioredis

기본 사용법

작업 큐 생성

import { Queue, Worker, QueueEvents } from 'bullmq';
import IORedis from 'ioredis';

const connection = new IORedis({ maxRetriesPerRequest: null });

const myQueue = new Queue('my-queue', { connection });
const myWorker = new Worker('my-queue', async (job) => {
  // 작업 처리 로직
  console.log(`작업 ${job.id} 처리 중...`);
  return { result: '성공' };
}, { connection });

const queueEvents = new QueueEvents('my-queue', { connection });

queueEvents.on('completed', ({ jobId }) => {
  console.log(`작업 ${jobId} 완료`);
});

queueEvents.on('failed', ({ jobId, failedReason }) => {
  console.error(`작업 ${jobId} 실패: ${failedReason}`);
});

작업 추가

// 단일 작업 추가
await myQueue.add('my-job', { foo: 'bar' });

// 지연된 작업 추가
await myQueue.add('delayed-job', { data: 'value' }, { delay: 10000 }); // 10초 지연

// 작업 추가 (재시도 설정 포함)
await myQueue.add('retry-job', { data: 'value' }, {
  attempts: 3,
  backoff: {
    type: 'exponential',
    delay: 2000
  }
});

// 여러 작업 추가 (벌크)
await myQueue.addBulk([
  { name: 'job-1', data: { id: 1 } },
  { name: 'job-2', data: { id: 2 } },
  { name: 'job-3', data: { id: 3 } },
]);

작업 처리

const myWorker = new Worker('my-queue', async (job) => {
  // 진행 상황 업데이트
  await job.updateProgress(50);
  
  // 작업 데이터에 접근
  const { foo } = job.data;
  
  // 작업이 취소되었는지 확인
  if (await job.isActive()) {
    throw new Error('작업이 취소됨');
  }
  
  // 작업 결과 반환
  return { processed: true };
});

오류 처리 및 재시도

const myWorker = new Worker('my-queue', async (job) => {
  try {
    // 위험한 작업 수행
    const result = await performRiskyOperation();
    return result;
  } catch (error) {
    // 재시도 여부를 결정하기 위해 오류 확인
    if (error instanceof TransientError) {
      // 재시도 허용 (attempts가 설정된 경우)
      throw error;
    } else {
      // 재시도 방지 (영구적 오류)
      await job.discard();
      throw error;
    }
  }
}, {
  connection,
  concurrency: 5, // 동시 작업 처리 수
});

작업 예약 (스케줄러)

import { QueueScheduler } from 'bullmq';

const scheduler = new QueueScheduler('my-queue', { connection });

// 반복 작업 추가
await myQueue.add('repeating-job', { type: 'cleanup' }, {
  repeat: {
    every: 3600000, // 1시간마다
    // cron 표현식 사용: '0 * * * *'
  }
});

고급 기능

BullMQ는 다음과 같은 고급 기능을 지원합니다:

• 샌드박스 워커: 별도 프로세스에서 작업 실행
• 작업 순서화: removeOnComplete, removeOnFail 옵션으로 큐 관리
• 작업 우선순위: 높은 우선순위 작업 먼저 실행
• 지연된 작업: 특정 시간 이후 실행
• 작업 그룹화: 관련 작업을 그룹으로 관리

모범 사례

1. 연결 재사용: 여러 BullMQ 인스턴스에서 동일한 IORedis 연결을 재사용하세요.
2. 오류 처리: 작업 핸들러 내에서 항상 try/catch 블록을 사용하세요.
3. 작업 크기 제한: 작업 데이터의 크기를 작게 유지하세요. 큰 데이터는 외부 저장소에 저장하고 참조하세요.
4. 모니터링: QueueEvents를 사용하여 시스템 상태를 추적하세요.
5. 정리: 오래된 작업은 자동 정리 옵션을 사용하여 큐를 깨끗하게 유지하세요.

참고 자료

• BullMQ 공식 문서
• BullMQ GitHub 저장소