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()));
}

Este ejemplo vemos cómo podemos utilizar worker_threads para realizar tareas en un hilo de trabajo separado y comunicarse con el hilo principal a través de mensajes.

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

En este ejemplo,

• El hilo principal (si isMainThread es verdadero) divide el rango de números (min a max) en sub-rangos para varios hilos de trabajo.
• Cada hilo de trabajo (Worker) recibe un sub-rango para calcular los números primos dentro de ese rango.
• Cuando un Worker encuentra números primos, envía esos primos de vuelta al hilo principal usando parentPort.postMessage.

Al ejecutar este script (node primeCalculator.js), verás que los números primos encontrados en cada sub-rango se imprimen en la consola.

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

Explicación del ejemplo,

• El hilo principal (si isMainThread es verdadero) carga las rutas de las imágenes que se van a procesar y divide estas rutas en subconjuntos para cada hilo de trabajo.
• Cada hilo de trabajo (Worker) recibe un conjunto de rutas de imágenes para procesar en paralelo.
• Dentro de cada hilo de trabajo, se carga cada imagen, se procesa y se guarda. El resultado final (la ruta de las imágenes procesadas) se envía de vuelta al hilo principal a través de mensajes.
• Al final, el hilo principal recibe los resultados de todos los hilos de trabajo y muestra el mensaje de procesamiento completo.

Descarga el código

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