Node.js Worker Threads Guide

Worker threads enable parallel JavaScript execution for CPU-intensive tasks. Here's how to use them effectively.

Basic Worker Thread#

// main.js
const { Worker, isMainThread, parentPort } = require('worker_threads');

if (isMainThread) {
  // Main thread
  const worker = new Worker(__filename);

  worker.on('message', (result) => {
    console.log('Result from worker:', result);
  });

  worker.on('error', (error) => {
    console.error('Worker error:', error);
  });

  worker.on('exit', (code) => {
    if (code !== 0) {
      console.error(`Worker exited with code ${code}`);
    }
  });

  worker.postMessage({ num: 42 });
} else {
  // Worker thread
  parentPort.on('message', (data) => {
    const result = heavyComputation(data.num);
    parentPort.postMessage(result);
  });
}

function heavyComputation(n) {
  let result = 0;
  for (let i = 0; i < 1000000000; i++) {
    result += Math.sqrt(n * i);
  }
  return result;
}

Separate Worker File#

// main.js
const { Worker } = require('worker_threads');
const path = require('path');

function runWorker(data) {
  return new Promise((resolve, reject) => {
    const worker = new Worker(path.join(__dirname, 'worker.js'), {
      workerData: data,
    });

    worker.on('message', resolve);
    worker.on('error', reject);
    worker.on('exit', (code) => {
      if (code !== 0) {
        reject(new Error(`Worker stopped with exit code ${code}`));
      }
    });
  });
}

// worker.js
const { workerData, parentPort } = require('worker_threads');

const result = processData(workerData);
parentPort.postMessage(result);

function processData(data) {
  // CPU-intensive work
  return data.map(item => item * 2);
}

// Usage
async function main() {
  const result = await runWorker([1, 2, 3, 4, 5]);
  console.log(result); // [2, 4, 6, 8, 10]
}

Worker Thread Pool#

const { Worker } = require('worker_threads');
const os = require('os');
const path = require('path');

class WorkerPool {
  constructor(workerPath, numWorkers = os.cpus().length) {
    this.workerPath = workerPath;
    this.numWorkers = numWorkers;
    this.workers = [];
    this.freeWorkers = [];
    this.taskQueue = [];

    this.initialize();
  }

  initialize() {
    for (let i = 0; i < this.numWorkers; i++) {
      this.addWorker();
    }
  }

  addWorker() {
    const worker = new Worker(this.workerPath);

    worker.on('message', (result) => {
      const { resolve } = worker.currentTask;
      resolve(result);
      this.freeWorkers.push(worker);
      this.processQueue();
    });

    worker.on('error', (error) => {
      const { reject } = worker.currentTask;
      reject(error);
      this.workers = this.workers.filter(w => w !== worker);
      this.addWorker();
    });

    this.workers.push(worker);
    this.freeWorkers.push(worker);
  }

  execute(data) {
    return new Promise((resolve, reject) => {
      this.taskQueue.push({ data, resolve, reject });
      this.processQueue();
    });
  }

  processQueue() {
    if (this.taskQueue.length === 0 || this.freeWorkers.length === 0) {
      return;
    }

    const worker = this.freeWorkers.pop();
    const task = this.taskQueue.shift();

    worker.currentTask = task;
    worker.postMessage(task.data);
  }

  async shutdown() {
    await Promise.all(
      this.workers.map(worker => worker.terminate())
    );
  }
}

// Usage
const pool = new WorkerPool('./worker.js', 4);

async function processItems(items) {
  const results = await Promise.all(
    items.map(item => pool.execute(item))
  );
  return results;
}

Shared Memory with SharedArrayBuffer#

// main.js
const { Worker } = require('worker_threads');

// Create shared buffer
const sharedBuffer = new SharedArrayBuffer(4 * 1024); // 4KB
const sharedArray = new Int32Array(sharedBuffer);

// Initialize data
for (let i = 0; i < sharedArray.length; i++) {
  sharedArray[i] = i;
}

// Create workers
const workers = [];
const numWorkers = 4;
const chunkSize = Math.ceil(sharedArray.length / numWorkers);

for (let i = 0; i < numWorkers; i++) {
  const worker = new Worker('./shared-worker.js', {
    workerData: {
      buffer: sharedBuffer,
      start: i * chunkSize,
      end: Math.min((i + 1) * chunkSize, sharedArray.length),
    },
  });
  workers.push(worker);
}

// Wait for all workers
Promise.all(
  workers.map(w => new Promise(resolve => w.on('message', resolve)))
).then(() => {
  console.log('Sum:', sharedArray.reduce((a, b) => a + b, 0));
});

// shared-worker.js
const { workerData, parentPort } = require('worker_threads');

const { buffer, start, end } = workerData;
const array = new Int32Array(buffer);

// Process assigned chunk
for (let i = start; i < end; i++) {
  array[i] = array[i] * 2;
}

parentPort.postMessage('done');

Atomics for Synchronization#

// Using Atomics for thread-safe operations
const { Worker, isMainThread, parentPort, workerData } = require('worker_threads');

if (isMainThread) {
  const sharedBuffer = new SharedArrayBuffer(4);
  const counter = new Int32Array(sharedBuffer);

  // Create multiple workers
  const workers = [];
  for (let i = 0; i < 4; i++) {
    const worker = new Worker(__filename, {
      workerData: { buffer: sharedBuffer },
    });
    workers.push(worker);
  }

  // Wait for all workers
  Promise.all(
    workers.map(w => new Promise(resolve => w.on('exit', resolve)))
  ).then(() => {
    console.log('Final count:', counter[0]); // Should be 4000000
  });
} else {
  const counter = new Int32Array(workerData.buffer);

  // Atomically increment 1 million times
  for (let i = 0; i < 1000000; i++) {
    Atomics.add(counter, 0, 1);
  }

  process.exit(0);
}

// Wait and notify
const { Worker, isMainThread, parentPort, workerData } = require('worker_threads');

if (isMainThread) {
  const buffer = new SharedArrayBuffer(4);
  const array = new Int32Array(buffer);

  const worker = new Worker(__filename, {
    workerData: { buffer },
  });

  setTimeout(() => {
    array[0] = 42;
    Atomics.notify(array, 0, 1); // Wake up one waiting thread
  }, 1000);

  worker.on('message', console.log);
} else {
  const array = new Int32Array(workerData.buffer);

  // Wait until notified
  Atomics.wait(array, 0, 0); // Wait while value is 0

  parentPort.postMessage(`Value is: ${array[0]}`);
}

Transferable Objects#

// Transfer ArrayBuffer ownership
const { Worker } = require('worker_threads');

const buffer = new ArrayBuffer(1024 * 1024); // 1MB

const worker = new Worker('./worker.js');

// Transfer ownership (zero-copy)
worker.postMessage(buffer, [buffer]);

// buffer is now detached (length === 0)
console.log(buffer.byteLength); // 0

// worker.js
const { parentPort } = require('worker_threads');

parentPort.on('message', (buffer) => {
  const array = new Uint8Array(buffer);
  // Process buffer...

  // Transfer back
  parentPort.postMessage(buffer, [buffer]);
});

// MessageChannel for bidirectional communication
const { Worker, MessageChannel } = require('worker_threads');

const { port1, port2 } = new MessageChannel();

const worker = new Worker('./channel-worker.js');
worker.postMessage({ port: port2 }, [port2]);

port1.on('message', (msg) => {
  console.log('Received:', msg);
});

port1.postMessage('Hello from main!');

// channel-worker.js
const { parentPort } = require('worker_threads');

parentPort.on('message', ({ port }) => {
  port.on('message', (msg) => {
    console.log('Worker received:', msg);
    port.postMessage('Hello from worker!');
  });
});

CPU-Intensive Task Examples#

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

if (isMainThread) {
  async function processImage(imageData) {
    return new Promise((resolve, reject) => {
      const worker = new Worker(__filename, {
        workerData: imageData,
      });
      worker.on('message', resolve);
      worker.on('error', reject);
    });
  }

  // Usage
  const imageBuffer = loadImage('photo.jpg');
  const processed = await processImage(imageBuffer);
} else {
  const imageData = workerData;

  // Apply filter
  const result = applyGrayscaleFilter(imageData);

  parentPort.postMessage(result);
}

// Cryptographic operations
const crypto = require('crypto');

// Hash computation in worker
if (isMainThread) {
  const pool = new WorkerPool('./hash-worker.js', 4);

  async function hashPasswords(passwords) {
    return Promise.all(
      passwords.map(pwd => pool.execute({ password: pwd }))
    );
  }
} else {
  parentPort.on('message', ({ password }) => {
    const hash = crypto.pbkdf2Sync(
      password,
      'salt',
      100000,
      64,
      'sha512'
    );
    parentPort.postMessage(hash.toString('hex'));
  });
}

// JSON parsing for large files
const { Worker } = require('worker_threads');
const fs = require('fs');

async function parselargeJSON(filePath) {
  return new Promise((resolve, reject) => {
    const worker = new Worker(`
      const { workerData, parentPort } = require('worker_threads');
      const fs = require('fs');

      const data = fs.readFileSync(workerData, 'utf8');
      const parsed = JSON.parse(data);

      parentPort.postMessage(parsed);
    `, {
      eval: true,
      workerData: filePath,
    });

    worker.on('message', resolve);
    worker.on('error', reject);
  });
}

Error Handling#

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

function createRobustWorker(workerPath, options = {}) {
  return new Promise((resolve, reject) => {
    const worker = new Worker(workerPath, options);

    const timeout = setTimeout(() => {
      worker.terminate();
      reject(new Error('Worker timed out'));
    }, options.timeout || 30000);

    worker.on('message', (result) => {
      clearTimeout(timeout);
      resolve(result);
    });

    worker.on('error', (error) => {
      clearTimeout(timeout);
      reject(error);
    });

    worker.on('exit', (code) => {
      clearTimeout(timeout);
      if (code !== 0) {
        reject(new Error(`Worker exited with code ${code}`));
      }
    });
  });
}

// Graceful shutdown
class ManagedWorkerPool {
  constructor(workerPath, size) {
    this.pool = new WorkerPool(workerPath, size);
    this.isShuttingDown = false;

    process.on('SIGTERM', () => this.shutdown());
    process.on('SIGINT', () => this.shutdown());
  }

  async execute(data) {
    if (this.isShuttingDown) {
      throw new Error('Pool is shutting down');
    }
    return this.pool.execute(data);
  }

  async shutdown() {
    this.isShuttingDown = true;
    await this.pool.shutdown();
    process.exit(0);
  }
}

Best Practices#

When to Use:
✓ CPU-intensive computations
✓ Image/video processing
✓ Large data parsing
✓ Cryptographic operations

Design:
✓ Use worker pools for reuse
✓ Transfer large data, don't copy
✓ Use SharedArrayBuffer carefully
✓ Handle errors and timeouts

Performance:
✓ Limit worker count to CPU cores
✓ Balance work distribution
✓ Minimize message passing
✓ Use Atomics for synchronization

Avoid:
✗ I/O-bound tasks in workers
✗ Creating workers for small tasks
✗ Sharing complex objects
✗ Race conditions without Atomics

Worker threads enable true parallelism for CPU-intensive tasks in Node.js. Use them for computations that would block the event loop. Implement worker pools for efficiency, SharedArrayBuffer for shared memory, and Atomics for synchronization. Remember that workers are best for CPU-bound work, not I/O operations.