Nest.js Bull

`Nest.js`는 강력한 백엔드 프레임워크로서, 복잡한 비즈니스 로직과 백그라운드 작업을 효과적으로 관리해야 할 때 사용할 수 있는 다양한 도구와 라이브러리를 제공합니다. 이 중에서도 `Bull.js`는 백그라운드 작업을 처리하기 위한 강력한 큐 라이브러리 중 하나로, `Nest.js`와 함께 사용하면 효율적인 작업 관리가 가능해집니다.

Bull.js란?

`Bull.js`는 `Node.js` 기반의 큐 라이브러리로, 백그라운드 작업을 비동기적으로 처리할 때 사용됩니다. `Bull.js`는 안정성과 성능을 중시하며, 다양한 기능과 옵션을 제공하여 다양한 상황에 적용할 수 있습니다.

 

 

Task

`Bull.js`에서 `task`는 큐(Queue)에 추가된 개별 작업을 나타냅니다. 이 작업은 비동기적으로 백그라운드에서 실행되어야 하는 일련의 작업 중 하나를 의미합니다. 예를 들어, 이메일 전송, 이미지 처리, 데이터 처리, 백엔드 작업 등 여러 종류의 작업이 큐에 추가될 수 있습니다.

 

작업(task)은 큐를 통해 비동기적으로 처리되기 때문에 큰 작업 부하를 주지 않고 여러 작업을 동시에 처리할 수 있으며, 실패한 작업을 재시도하거나 작업 상태를 모니터링하는 등의 기능을 제공합니다.

 

생명주기

`Bull.js`에서 작업(Task)의 생명주기는 작업이 큐에 추가되어 처리될 때까지의 과정을 설명합니다. 일반적으로 크게 4개의 stage로 나뉘며 자세하게는 다음과 같이 나뉩니다. 

 

 

작업 생성(Enqueue): 작업이 큐에 추가됩니다. 이 때 작업은 '대기(Waiting)' 상태로 들어갑니다. 대기 중인 작업은 실행을 위해 선택될 때까지 큐에 남아있습니다.

 

작업 선택(Active): 큐에서 작업 처리기(Worker)가 작업을 선택하면, 작업은 '활성(Active)' 상태가 됩니다. 이는 작업이 현재 처리 중임을 나타냅니다.

 

작업 완료(Completed): 작업이 성공적으로 완료되면, '완료(Completed)' 상태가 됩니다. 완료된 작업은 일반적으로 결과를 반환하고, 결과는 필요에 따라 저장하거나 다른 작업에 사용할 수 있습니다.

 

작업 실패(Failed): 작업 처리 중 오류가 발생하면, 작업은 '실패(Failed)' 상태가 됩니다. 실패한 작업은 일반적으로 로깅, 재시도, 오류 처리 등의 목적으로 사용됩니다.

 

작업 지연(Delayed): 작업은 설정된 시간만큼 지연되어 나중에 실행될 수 있습니다. 지연된 작업은 지정된 시간이 지나면 대기 상태로 돌아가 처리를 기다립니다.

 

작업 재시도(Retry): 실패한 작업은 설정에 따라 자동으로 재시도될 수 있습니다. 재시도 메커니즘을 사용하여 일시적인 오류나 네트워크 문제 등을 극복할 수 있습니다.

 

작업 제거(Remove): 작업이 완료되거나 특정 조건을 충족하면 큐에서 제거될 수 있습니다. 이는 시스템의 리소스를 관리하고, 큐를 깔끔하게 유지하는 데 도움이 됩니다.

 

작업 유지(Keep): 특정 설정에 따라, 작업이 완료된 후에도 큐에 남아있을 수 있습니다. 이는 추후 검토나 로깅 목적으로 사용될 수 있습니다.

 

상태 변화

각각의 Task의 상태를 변환하는 방법은 다음과 같은 조건을 따릅니다. 현재 태스크들의 상태와 수를 보고자 하면은 다음과 같은 함수를 작성하면 됩니다. 

const jobs = await this.testQueue.getJobCounts();

 

Waiting -> Active

`Waiting` 상태에 있는 데이터를 꺼내면은 자동으로 `Active`로 변경 됩니다.

const job = await this.testQueue.getNextJob();

 

Active -> Completed

`Completed` 하는 방법은 메소드를 통해 처리 가능합니다. 각각의 파라미터는 옵션을 나타내고 있습니다. 

await job.moveToCompleted("It's done!", true, true);

// moveToCompleted 설명
moveToCompleted(
  returnValue?: string, // 성공하면 반환 되는 값.
  ignoreLock?: boolean, // Queue Lock을 무시하고 성공으로 할지 여부
  notFetch?: boolean // 다음 Job을 자동으로 꺼낼지 여부 true시 수동으로 변경
): Promise<[any, JobId] | null>;

 

Active -> Failed

`Failed`로 메소드를 통해 처리 가능합니다. 

await job.moveToFailed({ message: error.message });

 

Consumer

`Consumer`는 `Queue`에 입력되 `Job`을 소비하는 클래스로 아래 사진과 같은 모양으로 데이터를 처리하고 있습니다. 데이터를 처리하는 방식도 간단하게 작성 할 수 있습니다. `getNextJob()` 함수를 사용하면 제일 먼저 처리 되어져야 하는 `job`이 나오고 처리를 하면 됩니다. 

await this.testQueue.add('test-job-1', { data });
await this.testQueue.add('test-job-2', { data });

 

 

각 `Job`은 `name`을 지정할 수 있는데 `name`에 맞게 `Process` 지정하여 자동으로 처리하도록 할 수 있습니다.

const job = await this.testQueue.getNextJob();

 

 

 

 

Processor

`Processor`는 `Bull.js`에서 작업을 처리하는 클래스 데코레이터입니다. 작업 처리를 담당하는 클래스에 `@Processor(queueName)` 데코레이터를 사용하여 해당 클래스가 어떤 큐에서 작업을 처리할지를 정의합니다.

import { Processor } from 'bullmq';

@Processor('myQueue')
class MyQueueProcessor {
  // 작업 처리 로직 구현
}

 

Process

`Process` 데코레이터는 `Processor` 클래스 내에서 실제 작업을 처리하는 메서드에 적용됩니다. 이 데코레이터는 해당 메서드가 어떤 작업을 처리해야 하는지를 정의합니다.

import { Process } from 'bullmq';

@Processor('myQueue')
class MyQueueProcessor {
  @Process('name')
  async handleJob(job: Job) {
    // 작업을 처리하는 로직을 여기에 구현
  }
}

 

 

 

주의사항

• `add`시에  `jobId`를  지정이 가능한데 이때 `queue` 에서 제공하는 `id` 를 사용하지 않을 경우, `getNextJob()` 함수를 사용 불가 합니다. 
• `completed` 와 `failed` 상태로 변경하고자 하면 제공되는 함수를 통해 가능합니다.