Cómo trabajar con Threads en Node.js

Los hilos de trabajo (threads) proporcionan una forma de ejecutar código de forma concurrente en Node.js, lo que permite realizar tareas intensivas en CPU de manera eficiente y sin bloquear el hilo principal de ejecución.

El módulo worker_threads en Node.js proporciona una API para crear y administrar hilos de trabajo (threads). Cada hilo de trabajo ejecuta su propio contexto de JavaScript y se comunica con el hilo principal a través de mensajes.

Ejemplos de uso del módulo worker_threads

Ejemplo básico

Comencemos con un ejemplo básico de cómo utilizar worker_threads para realizar tareas en paralelo:

import { Worker, isMainThread, workerData, parentPort } from 'node:worker_threads';

// Verifica si el hilo actual es el hilo principal
if (isMainThread) {
  // Si es el hilo principal, se define un conjunto de datos
  const data = 'some data';
  // Se crea un nuevo hilo de trabajo y se le pasa los datos definidos
  const worker = new Worker(import.meta.filename, { workerData: data });
  // Se escucha por mensajes enviados desde el hilo de trabajo
  worker.on('message', msg => console.log('Reply from Thread:', msg));
} else {
  // Si no es el hilo principal, se obtienen los datos pasados al hilo de trabajo
  const source = workerData;
  // Se convierte el texto a mayúsculas y luego se codifica a base64
  parentPort.postMessage(btoa(source.toUpperCase()));
}

Ejemplo: Cálculo de números primos en paralelo

Supongamos que queremos calcular los números primos en un rango grande de números de manera eficiente utilizando hilos de trabajo.

Podemos dividir el rango en varios sub-rangos y asignar un hilo de trabajo para calcular los primos en cada sub-rango.

// primeCalculator.js
import { Worker, isMainThread, parentPort, workerData } from 'node:worker_threads';

function isPrime(num) {
  if (num <= 1) return false;
  if (num <= 3) return true;
  if (num % 2 === 0 || num % 3 === 0) return false;
  let i = 5;
  while (i * i <= num) {
    if (num % i === 0 || num % (i + 2) === 0) return false;
    i += 6;
  }
  return true;
}

if (isMainThread) {
  const min = 2;
  const max = 100000;
  const numThreads = 4;
  const range = Math.ceil((max - min) / numThreads);

  for (let i = 0; i < numThreads; i++) {
    const start = min + range * i;
    const end = i === numThreads - 1 ? max : start + range;
    const worker = new Worker(__filename, {
      workerData: { start, end },
    });
    worker.on('message', (message) => {
      console.log('Primes found:', message.primes);
    });
  }
} else {
  const { start, end } = workerData;
  const primes = [];
  for (let i = start; i < end; i++) {
    if (isPrime(i)) {
      primes.push(i);
    }
  }
  parentPort.postMessage({ primes });
}

Ejemplo: Procesamiento de imágenes en paralelo

Supongamos que tenemos una aplicación que necesita procesar una gran cantidad de imágenes de manera eficiente.

Podemos utilizar hilos de trabajo para distribuir el procesamiento de imágenes en paralelo, lo que aceleraría significativamente el tiempo de procesamiento.

import { Worker, isMainThread, parentPort, workerData } from 'node:worker_threads';
import { loadImage, processImage, saveImage } from './imageUtils.js';

if (isMainThread) {
  const imagePaths = [
    'image1.jpg',
    'image2.jpg',
    'image3.jpg',
    // Agregar más rutas de imágenes según sea necesario
  ];

  const numThreads = 4;
  const imagesPerThread = Math.ceil(imagePaths.length / numThreads);

  for (let i = 0; i < numThreads; i++) {
    const start = i * imagesPerThread;
    const end = start + imagesPerThread;
    const paths = imagePaths.slice(start, end);

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

    worker.on('message', ({ processedImages }) => {
      console.log('Procesamiento de imágenes completo:', processedImages);
    });
  }
} else {
  const { paths } = workerData;
  const processedImages = [];

  paths.forEach(async (path) => {
    try {
      const image = await loadImage(path);
      const processedImage = await processImage(image);
      const savedPath = await saveImage(processedImage);
      processedImages.push(savedPath);
    } catch (error) {
      console.error('Error al procesar la imagen:', error);
    }
  });

  parentPort.postMessage({ processedImages });
}

Descarga el código

Todo el código de esta entrada está disponible para su descarga en Github