Node.js: Worker Threads

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

const worker = new Worker('./heavy-task.js', {
  workerData: { taskId: 42, input: 'hello' }
});

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

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

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

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

const { taskId, input } = workerData;

const result = performHeavyComputation(input);

parentPort.postMessage({ taskId, result });

الدالة performHeavyComputation(data) {
  let hash = 0;
  for (let i = 0; i < 1e8; i++) {
    hash = (hash + data.charCodeAt(i % data.length)) % 65536;
  }
  return hash;
}

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

const worker = new Worker(`
  const { parentPort } = require('worker_threads');
  parentPort.postMessage('Hello from worker!');
`, { eval: true });

worker.on('message', (msg) => {
  console.log(msg); // Hello from worker!
});

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

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

worker.postMessage({ action: 'greet', name: 'Alice' });

worker.on('message', (msg) => {
  console.log('Main received:', msg);
  if (msg.action === 'greet_reply') {
    worker.postMessage({ action: 'task', payload: 100 });
  }
});

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

parentPort.on('message', (msg) => {
  if (msg.action === 'greet') {
    parentPort.postMessage({
      action: 'greet_reply',
      text: `Hello, ${msg.name}!`
    });
  } else if (msg.action === 'task') {
    const result = msg.payload * 2;
    parentPort.postMessage({ action: 'result', القيمة: result });
  }
});

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

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

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

worker.postMessage({ buffer }, [buffer]);

console.log('Buffer transferred, main thread no longer owns it');

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

parentPort.on('message', ({ buffer }) => {
  const view = new Uint8Array(buffer);
  view[0] = 42;
  parentPort.postMessage({ done: true, firstByte: view[0] }, [buffer]);
});

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

const imageFiles = [
  'photo-001.jpg', 'photo-002.jpg', 'photo-003.jpg',
  'photo-004.jpg', 'photo-005.jpg', 'photo-006.jpg'
];

const WORKER_COUNT = 4;
const chunkSize = Math.ceil(imageFiles.length / WORKER_COUNT);

for (let i = 0; i < WORKER_COUNT; i++) {
  const chunk = imageFiles.slice(i * chunkSize, (i + 1) * chunkSize);
  const worker = new Worker('./image-worker.js', {
    workerData: {
      workerId: i,
      files: chunk,
      outputDir: './thumbnails'
    }
  });

  worker.on('message', (result) => {
    console.log(`Worker ${i} done:`, result.processed);
  });
}

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

const { workerId, files, outputDir } = workerData;

async function generateThumbnail(file) {
  // Simulate image processing
  return new Promise((resolve) => {
    setTimeout(() => resolve(`${file} -> thumb`), 100);
  });
}

(async () => {
  const processed = [];
  for (const file of files) {
    const result = await generateThumbnail(file);
    processed.push(result);
  }
  parentPort.postMessage({ workerId, processed });
})();

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

const worker1 = new Worker('./chat-worker.js', {
  workerData: { id: 1 }
});
const worker2 = new Worker('./chat-worker.js', {
  workerData: { id: 2 }
});

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

worker1.postMessage({ port: port1 }, [port1]);
worker2.postMessage({ port: port2 }, [port2]);

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

parentPort.once('message', ({ port }) => {
  port.on('message', (msg) => {
    console.log(`Worker ${workerData.id} received:`, msg);
  });

  setInterval(() => {
    port.postMessage(`Hello from Worker ${workerData.id}`);
  }, 1000);
});

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

const sharedBuffer = new SharedArrayBuffer(4);
const sharedArray = new Int32Array(sharedBuffer);

const WORKER_COUNT = 4;
const INCREMENTS_PER_WORKER = 100000;

for (let i = 0; i < WORKER_COUNT; i++) {
  const worker = new Worker('./counter-worker.js', {
    workerData: { sharedBuffer, increments: INCREMENTS_PER_WORKER }
  });

  worker.on('exit', () => {
    const final = Atomics.load(sharedArray, 0);
    console.log(`Final counter value: ${final}`);
  });
}

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

const { sharedBuffer, increments } = workerData;
const sharedArray = new Int32Array(sharedBuffer);

for (let i = 0; i < increments; i++) {
  Atomics.add(sharedArray, 0, 1);
}

parentPort.postMessage('done');

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

class Pool {
  constructor(workerFile, size) {
    this.workerFile = workerFile;
    this.size = size;
    this.workers = [];
    this.queue = [];

    for (let i = 0; i < size; i++) {
      const worker = new Worker(workerFile);
      worker.busy = false;
      worker.on('message', (result) => {
        const task = worker.currentTask;
        worker.busy = false;
        worker.currentTask = null;
        task.resolve(result);
        this._processQueue();
      });
      worker.on('error', (err) => {
        const task = worker.currentTask;
        if (task) {
          worker.busy = false;
          worker.currentTask = null;
          task.reject(err);
          this._processQueue();
        }
      });
      this.workers.push(worker);
    }
  }

  run(data) {
    return new Promise((resolve, reject) => {
      const task = { data, resolve, reject };
      const idle = this.workers.find((w) => !w.busy);
      if (idle) {
        this._assign(idle, task);
      } else {
        this.queue.push(task);
      }
    });
  }

  _assign(worker, task) {
    worker.busy = true;
    worker.currentTask = task;
    worker.postMessage(task.data);
  }

  _processQueue() {
    if (this.queue.length === 0) return;
    const idle = this.workers.find((w) => !w.busy);
    if (!idle) return;
    const task = this.queue.shift();
    this._assign(idle, task);
  }

  destroy() {
    for (const worker of this.workers) {
      worker.terminate();
    }
    this.workers = [];
    this.queue = [];
  }
}

module.exports = Pool;

// main.js
const Pool = require('./pool.js');

const pool = new Pool('./fib-worker.js', 4);

async function main() {
  const tasks = [40, 41, 42, 43, 44, 45, 46, 47];

  const promises = tasks.map((n) => pool.run({ n }));

  const results = await Promise.all(promises);

  for (let i = 0; i < tasks.length; i++) {
    console.log(`fib(${tasks[i]}) = ${results[i]}`);
  }

  pool.destroy();
}

main();

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

parentPort.on('message', ({ n }) => {
  const result = fib(n);
  parentPort.postMessage(result);
});

function fib(n) {
  if (n <= 1) return n;
  let a = 0, b = 1;
  for (let i = 2; i <= n; i++) {
    [a, b] = [b, a + b];
  }
  return b;
}

npm install piscina

const path = require('path');
const Piscina = require('piscina');

const pool = new Piscina({
  filename: path.resolve(__dirname, 'task.js'),
  maxThreads: 4
});

async function main() {
  const results = await Promise.all([
    pool.run({ x: 10, y: 20 }),
    pool.run({ x: 30, y: 40 }),
    pool.run({ x: 50, y: 60 })
  ]);

  console.log(results); // [30, 70, 110]
  await pool.destroy();
}

main();

// task.js
module.exports = ({ x, y }) => {
  return x + y;
};

// image-pool.js
const { Worker } = require('worker_threads');
const path = require('path');

class ImagePool {
  constructor(workerCount) {
    this.workers = [];
    this.queue = [];
    this.results = [];

    for (let i = 0; i < workerCount; i++) {
      const worker = new Worker(path.join(__dirname, 'image-processor.js'));
      worker.busy = false;

      worker.on('message', (msg) => {
        if (msg.type === 'result') {
          this.results.push(msg.data);
        }
        worker.busy = false;
        worker.currentResolve();
        this._dispatch();
      });

      worker.on('error', (err) => {
        worker.busy = false;
        if (worker.currentReject) {
          worker.currentReject(err);
        }
        this._dispatch();
      });

      this.workers.push(worker);
    }
  }

  process(fileList) {
    this.results = [];
    const promises = fileList.map((file) => this._enqueue(file));
    return Promise.all(promises).then(() => this.results);
  }

  _enqueue(file) {
    return new Promise((resolve, reject) => {
      this.queue.push({ file, resolve, reject });
      this._dispatch();
    });
  }

  _dispatch() {
    while (this.queue.length > 0) {
      const idle = this.workers.find((w) => !w.busy);
      if (!idle) break;
      const task = this.queue.shift();
      idle.busy = true;
      idle.currentResolve = task.resolve;
      idle.currentReject = task.reject;
      idle.postMessage({ file: task.file, sizes: [200, 400, 800] });
    }
  }

  destroy() {
    this.workers.forEach((w) => w.terminate());
  }
}

module.exports = ImagePool;

// image-processor.js
const { parentPort } = require('worker_threads');

parentPort.on('message', async ({ file, sizes }) => {
  const results = [];

  for (const size of sizes) {
    const thumb = await resize(file, size);
    results.push(thumb);
  }

  parentPort.postMessage({
    type: 'result',
    data: { file, thumbnails: results }
  });
});

async function resize(file, maxSize) {
  return new Promise((resolve) => {
    const duration = Math.random() * 200 + 50;
    setTimeout(() => {
      resolve(`${file}_${maxSize}px.jpg`);
    }, duration);
  });
}

// run.js
const ImagePool = require('./image-pool.js');

async function main() {
  const pool = new ImagePool(4);

  const files = Array.from({ length: 20 }, (_, i) =>
    `photo-${String(i + 1).padStart(3, '0')}.jpg`
  );

  const start = Date.now();
  const results = await pool.process(files);
  const elapsed = Date.now() - start;

  console.log(`Processed ${files.length} images in ${elapsed} ms`);
  console.log(`Generated ${results.reduce((sum, r) => sum + r.thumbnails.length, 0)} thumbnails`);

  pool.destroy();
}

main();

node run.js

Processed 20 images in 1234 ms
Generated 60 thumbnails