[NestJS] Worker Thread를 사용해봅시다.

안녕하세요 formulous입니다.

 

이번에 제가 구성한 NestJS 서버가  TIME_OUT 응답을 종종 주더라구요.

 

원인을 찾아보니, NestJS의 메인 스레드가 대용량의 데이터 작업으로 인해 다른 요청을 수신할 수 없는 상태라는 것을 발견하게 되었어요.

 

이처럼 NestJS의 근간이 되는 Nodejs는 싱글 스레드로 실행되기 때문에, 대용량의 작업을 실행 중이라면 다른 작업을 동시에 진행할 수 없게 되는 문제가 있어요.

 

하지만 늘 그렇듯 해결법은 존재합니다.

 

 

 

대용량 작업에 의한 싱글스레드 블로킹을 방지할 수 있는 워커 스레드(Worker Thread)에 대해 알아보겠습니다.

 

워커 스레드를 사용하기 전 개념과 사용하는 이유에 대해서 알아볼까요?

 

우선 개념적으로 말씀드리자면, 워커 스레드는 NodeJS에서 멀티 스레드 프로그래밍을 가능하게 해주는 기능입니다.

 

쉽게 말해서 일꾼(Worker)에게 일을 맡겨두고 주인장은 다른 일을 처리할 수 있는 상태로 유지한다는 거죠.

 

또한, 메인스레드와 독립적인 메모리 공간을 사용하므로 워커 스레드에서 예외 상황이 발생하는 경우 메인 스레드에 별도 영향이 없다는 것도 큰 장점이에요.

 

하지만 아쉽게도 워커 스레드에도 단점이 존재해요.

 

 

워커 스레드는 위와 같은 CPU 연산 작업에는 유용하게 사용할 수 있지만 Database 관련 작업과 파일 읽기/쓰기와 같은 I/O 작업에서는 큰 효율을 발휘하지 않아요.

(I/O 작업의 경우 Promise, async/await 방식을 권장해요)

 

워커스레드는 메인스레드와 데이터를 주고받을 때 반드시 데이터를 JSON 형태로 변환하고 다시 파싱 하는 과정이 필요한데, 불필요한 곳에 워커스레드를 생성하여 사용하는 경우 데이터 전송 오버헤드가 발생할 수도 있죠.

 

또한, 스레드 간 통신이 필요한 경우 디버깅이 상당히 복잡해지기에 대량의 CPU 연산이 메인 스레드 동작에 지장을 주는 경우에 한해서 사용하는 게 좋아요.

 

 

자, 이제 본론으로 들어가서 NestJS 서버에서 워커스레드를 어떤 방식으로 적용하는지 알아볼까요?

 

우선, 워커 스레드를 실행하는 방법은 두 가지가 있어요.

 

• 개별 worker.js 파일을 만들어 실행한다.
• inline 방식으로 코드 내 인스턴스를 생성하여 실행한다.

 

통상적으로는 첫 번째 방법을 많이 쓰게 되는데요.

 

우선, 워커 스레드에 쓰일 js 파일을 작성해 볼까요?

 

// **/worker_path/worker.js

const { parentPort } = require('worker_threads');

// requestData는 NestJS에서 보내주는 데이터 입니다.
parentPort.on('message', (requestData) => {
  // 내부 작업을 해당 Callback 함수 내에 정의해줍니다.
  try {
  
    // 성공 응답을 전달해요.
    parentPort.postMessage('Worker received: ' + requestData);
    
  } catch (err) {
  
    // 실패 응답을 전달해요.
    throw new Error(JSON.stringify(err));
    
  } finally {
    parentPort.close();
  }
});

 

 

requestData는 node 서버에서 워커 스레드로 전달해 주는 데이터로 parentPort.on의 콜백함수 내에  해당 데이터를 활용하여 실행할 작업을 기재해 주면 돼요.

 

이제 NestJS에서 어떤 방식으로 데이터를 전달하는지 확인해 볼게요.

 

import { Worker } from 'worker_threads';

// 워커 스레드 파일과 연결하여, 워커스레드의 응답을 대기할 수 있도록 Promise 객체로 응답해주고 있어요.
function processWithWorker(requestData): Promise<any> {
  return new Promise((resolve, reject) => {
    // 워커스레드 파일과 연결하여 워커스레드 인스턴스를 생성해요.
    const worker = new Worker('worker_path/worker.js');
    
    // 워커스레드에 요청 데이터를 전송해요.
    worker.postMessage(requestData);
    
    // 워커 스레드 응답 이벤트를 리스닝하고 있어요.
    // 워커 스레드에서 보낸 응답이 data 변수에 담겨와요.
    worker.on('message', (data) => {
      resolve(data);
    });
    
    // 워커 스레드에서 에러가 발생한 경우 이벤트에 리스닝하고있어요.
    worker.on('exit', (data) => {
      if (data === 1) {
        reject(null);
      }
    });
    
  });
}

async () => {
  const workerReponse = await processWithWorker('TEST DATA')
}();

 

 

오늘은 간단하게 워커 스레드의 개념과, 그 활용 방법에 대해서 알아보았어요.

 

대량의 CPU 연산 작업이 있다면 위 개념을 참고해서 유용하게 활용해 보셨으면 좋겠어요.

 

감사합니다.