Node.js | worker_threads module

trevor1107·2021년 2월 18일

worker_threads module

노드에서 멀티 스레드 방식으로 작업하게 해주는 모듈이다.

간단하게 사용법을 익혀보자

const workerThreads = require('worker_threads');

// mainThread
if (workerThreads.isMainThread) {
    // workerThread 생성(대상)
    const worker = new workerThreads.Worker(__filename);
    // 작업 준비(메세지) 메세지가 오면 실행
    worker.on('message', (message) => console.log('from worker :', message));
    // 작업 준비(종료) 종료지시가 오면 실행
    worker.on('exit', () => console.log('worker exit'));
    // 부모 스레드에게 데이터 메세지를 보낸다.
    worker.postMessage('ping');
}
// workerThread 실행 작업문
else {
    // 부모 스레드 작업 준비(메세지)
    // 메세지가 오면 실행한다.
    workerThreads.parentPort.on('message', (value) => {
        // 받은 메세지를 표시
        console.log('from parent :', value);
        // 워커 스레드에게 데이터 메세지를 보낸다.
        workerThreads.parentPort.postMessage('pong');
        // 부모 스레드와 자식 스레드의 연결을 종료시킨다.
        // 워커 스레드의 worker.on('exit')가 실행된다.
        workerThreads.parentPort.close();
    });
    console.log('workerThread logic');
}

결과)

workerThread logic
from parent : ping
from worker : pong
worker exit

여러개의 워크 스레드를 실행해보자

const workerThreads = require('worker_threads');
const Worker = workerThreads.Worker;
if (workerThreads.isMainThread) {
    // 자료구조 생성
    const threads = new Set();

    // 워크 스레드 추가
    threads.add(
        new Worker(__filename, {
            workerData: { start: 1 },
        })
    );
    threads.add(
        new Worker(__filename, {
            workerData: { start: 5 },
        })
    );
    // 모든 워크 스레드를 순회
    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('worker done');
        });
    }
} else {
    // 실제 워커 쓰레드가 실행하는 구문, 데이터 메세지를 보낸다.
    const data = workerThreads.workerData;
    workerThreads.parentPort.postMessage(data.start + 100);
}

결과)

from worker : 101
from worker : 105
worker done

반복문을 통한 싱글 스레드와 멀티 스레드 비교

2부터 100만까지의 숫자중에 소수가 모두 몇 개 있는지를 알아내는 로직이다.

워커 스레드를 사용하지 않는 싱글 스레드의 경우

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

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

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

결과)

prime: 약 490.ms
length: 9592

워커 스레드를 사용하는 멀티 스레드의 경우

const workerThreads = require('worker_threads');
const Worker = workerThreads.Worker;
const workerData = workerThreads.workerData;
const min = 2;
let primes = [];

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

if (workerThreads.isMainThread) {
    const max = 1000000;
    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('length :', primes.length);
            }
        });
        worker.on('message', (msg) => {
            primes = primes.concat(msg);
        });
    }
} else {
    findPrimes(workerData.start, workerData.range);
    workerThreads.parentPort.postMessage(primes);
}