worker_threads

worker_threads.js

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

if (isMainThread) {
  // 부모일 때
  const worker = new Worker(__filename);
  worker.on("message", (message) => console.log("from worker", message));
  worker.on("exit", () => console.log("worker exit"));
  worker.postMessage("ping");
} else {
  // 워커일 때
  parentPort.on("message", (value) => {
    console.log("from parent", value);
    parentPort.postMessage("pong");
    parentPort.close();
  });
}

 

isMainThread를 통해 현재 코드가 메인 스레드(기존에 동작하던 싱글 스레드를 메인 스레드 또는 부모 스레드하고 부릅니다)에서 실행되는지, 아니면 우리가 생성한 워커 스레드에서 실행되는지 구분됩니다. 메인 스레드에서는 new Worker를 통해 현재 파일(__filename)을 워커 스레드에서 실행시키고 있습니다. 물론 현재 파일의 else 부분민 워커 스레드에서 실행됩니다.

부모에서는 워커 생성 후 worker,postMessage로 워커에 데이터를 보낼 수 있습니다. 워커는 paraentPort.on('message') 이벤트 리스너로 부모로부터 메시지를 받고, parentPort.postMessage로 부모에게 메시지를 보냅니다. 부모는 worker.on('message')로 메시지를 받습니다. 참고로 메시지를 한 번만 받고 싶다면 once('message')를 사용하면 됩니다.

워커에서 on 메서드를 사용할 때는 직접 워커를 종료해야 한다는 점에 주의하세요. parentPort.close()를 하면 부모와의 연결이 종료됩니다. 종료될 때는 worker.on('exit')이 실행됩니다.

 

콘솔

$ node woker_threads
from parent ping
from worker pong
worker exit

 

메인 스레드와 워커의 통신

 

worker_data.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);
}

 

new Worker를 호출할 때 두 번째 인수의 workerData 속성으로 원하는 데이터를 보낼 수 있습니다. 워커에서는 workerData로 부모로부터  데이터를 받습니다. 현재 두 개의 워커가 돌이가고 있으며, 각각 부모로부터 숫자를 받아서 100을 더해 돌려줍니다. 돌려주는 순간 워커가 종료되어 worker.on('exit')이 실행됩니다. 워커 두개가 모두 종료되면 job done이 로깅됩니다.

 

콘솔

$ node worker_data
from worker 101
from worker 102
job done

 

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부터 1,000만까지의 숫자 중에 소수가 모두 몇 개 있는지를 알아내는 코드입니다.

 

콘솔

$ node prime
prime: 8.495s
664579

 

사용자의 컴퓨터 성능에 따라 다르지만 상당한 시간이 소요됩니다. 이번에는 워커 스레드를 사용하여 여러 개의 스레드들이 문제를 나눠서 풀도록 해보겠습니다.

 

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.ceil((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: range + ((max - min + 1) % threadCount) },
    })
  );
  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);
}

 

여덟 개의 스레드가 일을 나눠서 처리했습니다. 멀티 스레딩을 할 때는 일을 나눠서 처리하도록 하는 게 제일 어렵습니다. 어떠한 일은 공유하고 있는 데이터가 많아 일일 나누기가 어렵습니다. 다행히 소수의 개수를 구하는 작업은 정해진 범위(2부터 1,000만)를 스레드들이 일정하게 나눠서 수행할 수 있습니다.

 

콘솔

$ node prime-worker
prime: 1.856s
664579

 

속도가 여섯 배 정도 빨라졌습니다. 워커 스레드를 여덟 개 사용했다고 해서 여덟 배 빨라지는 것은 아닙니다. 스레드를 생성하고 스레드 사이에서 통신하는 데 상당한 비용이 발생하므로, 이 점을 고려해서 멀티 스레딩을 해야 합니다. 잘못하면 멀티 스레딩을 할 때 싱글 스레딩보다 더 느려지는 현상도 발생할 수 있습니다.