Node.js 내장 모듈 - worker_threads

오늘은 Node.js 내장 모듈인 worker_threads에 대해 알아볼 것이다.

(노드에서 worker_threads를 쓰는 경우는 극히 드물다..!)

(멀티 스레드로 작업하는게 쉬운 일은 아니다.)

노드에서 멀티 스레드 방식으로 작업하는 경우 사용한다. (cpu를 많이 잡아먹는 암호화나 압축 등에 사용)

• isMainThread : 현재 코드가 메인 스레드에서 실행되는지, 워커 스레드에서 실행되는지 구분.
• 메인 스레드에서는 new Worker를 통해 현재 파일(__filename)을 워커 스레드에서 실행시킴.
• worker.postMessage로 부모에서 워커로 데이터를 보냄.
• parentPort.on('message')로 부모로부터 데이터를 받고, postMessage로 데이터를 보냄

사용 방법

worker_threads.js

const {
  Worker, isMainThread, parentPort, workerData,
} = require('worker_threads');

if (isMainThread) { // 부모일 때
  const threads = new Set();
  threads.add(new Worker(__filename, {
    workerData: { start: 1 },
  }));
  threads.add(new Worker(__filename, {
    workerData: { start: 2 },
  }));
  for (let worker of threads) {
    worker.on('message', message => console.log('from worker', message));
    worker.on('exit', () => {
      threads.delete(worker);
      if (threads.size === 0) {
        console.log('job done');
      }
    });
  }
} else { // 워커일 때
  const data = workerData;
  parentPort.postMessage(data.start + 100);
}

콘솔 결과

worker_threads.js 코드는 워커 스레드를 쓰는 방식을 간단하게 보여준 것이다.

 

그렇다면 정확히 어떤 부분에서 사용하면 좋을 지 소수를 찾는 예제 코드를 먼저 보자.

아래 prime.js 코드는 에라토스테네스의 체 알고리즘을 싱글 스레드로 구현한 것이다.

 

prime.js

const min = 2;
const max = 10000000;
const primes = [];

function findPrimes(start, range) {
  let isPrime = true;
  const end = start + range;
  for (let i = start; i < end; i++) {
    for (let j = min; j < Math.sqrt(end); j++) {
      if (i !== j && i % j === 0) {
        isPrime = false;
        break;
      }
    }
    if (isPrime) {
      primes.push(i);
    }
    isPrime = true;
  }
}

console.time('prime');
findPrimes(min, max);
console.timeEnd('prime');
console.log(primes.length);

콘솔 결과

싱글 스레드로 코드를 짜게 된다면 2 부터 10,000,000까지 구하는데 걸리는 시간은 4.579초가 걸리게 된다.

만약 여러 사람이 요청 한다면 뒤로 갈수록 더욱 더 오래 걸릴 것이다.

 

그렇다면 워커 스레드로 코드를 짜보자.

 

prime-worker.js

const { Worker, isMainThread, parentPort, workerData } = require('worker_threads');

const min = 2;
let primes = [];

function findPrimes(start, range) {
  let isPrime = true;
  const end = start + range;
  for (let i = start; i < end; i++) {
    for (let j = min; j < Math.sqrt(end); j++) {
      if (i !== j && i % j === 0) {
        isPrime = false;
        break;
      }
    }
    if (isPrime) {
      primes.push(i);
    }
    isPrime = true;
  }
}

if (isMainThread) {
  const max = 10000000;
  const threadCount = 8;
  const threads = new Set();
  const range = Math.floor((max - min) / threadCount);
  let start = min;
  console.time('prime');
  for (let i = 0; i < threadCount - 1; i++) {
    const wStart = start;
    threads.add(new Worker(__filename, { workerData: { start: wStart, range } }));
    start += range;
  }
  threads.add(new Worker(__filename, { workerData: { start, range: max - start } }));
  for (let worker of threads) {
    worker.on('error', (err) => {
      throw err;
    });
    worker.on('exit', () => {
      threads.delete(worker);
      if (threads.size === 0) {
        console.timeEnd('prime');
        console.log(primes.length);
      }
    });
    worker.on('message', (msg) => {
      primes = primes.concat(msg);
    });
  }
} else {
  findPrimes(workerData.start, workerData.range);
  parentPort.postMessage(primes);
}

콘솔 결과

싱글 스레드로 코드를 짰을 때 4.579초가 걸리던게 워커 스레드를 8개로 만들어서 일을 분배하니 1.088초로 약 5배가 줄어들었다.

(일을 분배할 때 2~10,000,000까지 숫자를 나눠서 분배해야 한다.)

워커 스레드를 8개로 만들었는데 시간이 8분의 1로 줄어들지 않는 이유는 워커 스레드를 많이 늘린다고 해서 시간이 비례해서 줄어들지 않는다.
그리고 내 컴퓨터 코어는 5개인데 워커 스레드가 8개면 5개가 먼저 돌아간 뒤 남는 3개는 5개가 끝나고 돌아간다.
결론은 스레드 개수를 수정을 하면서 적절한 숫자를 찾아야 한다.

실무에서는 워커에서 에러 났을 때 복구 코드까지 작성해야 한다.

(사실 Node로는 멀티 스레드를 하는 것을 추천하지 않고 다른 언어로 하는 것을 추천한다..)