En el modelo de concurrencia mono-hilo de Node.js, ejecutar cómputos intensivos de CPU (como el procesamiento de archivos JSON de varios gigabytes o la transformación de vídeos) o tareas pesadas de entrada/salida por lotes (como la generación de informes en CSV/XLSX con 500.000 filas o el envío de miles de webhooks) directamente en el ciclo de vida de una petición HTTP resulta inaceptable. Dicha práctica bloquea el Event Loop, dispara el retraso en la atención de eventos (Event Loop Delay) e imposibilita la admisión de nuevas conexiones de red.
La solución de nivel de producción consiste en delegar el trabajo asíncrono a colas de mensajes distribuidas. BullMQ es una biblioteca empresarial de gestión de colas para Node.js y TypeScript construida sobre las estructuras de datos nativas de Redis. A diferencia de los intermediarios de mensajería tradicionales (RabbitMQ, Apache Kafka), BullMQ combina el procesamiento ultrarrápido en memoria de Redis con capacidades avanzadas de planificación: ejecuciones diferidas, programaciones cron, prioridades numéricas, limitación de frecuencia (Rate Limiting) y flujos jerárquicos basados en Grafos Acíclicos Dirigidos (DAG).
BullMQ no implementa colas mediante bucles de sondeo abstractos; opera directamente sobre las primitivas de Redis mediante scripts Lua precompilados y atómicos:
QueueEvents para propagar eventos del ciclo de vida entre workers y clientes observadores con mínimo consumo de CPU y memoria.zset): Gestionan tareas diferidas (delayed), tareas recurrentes (repeatable) y tareas con prioridad (prioritized). La puntuación (score) almacena marcas de tiempo Unix en milisegundos o valores numéricos de prioridad.hash): Contienen los metadatos específicos de cada trabajo (job:id): carga útil (payload), intentos efectuados (attemptsMade), trazas de error (stacktrace), progreso y valor devuelto.list): Configuran las colas de espera FIFO/LIFO (wait, paused) y los registros de tareas completadas o fallidas. El desplazamiento atómico entre listas se ejecuta mediante las instrucciones RPOPLPUSH / LMOVE.La arquitectura de un subsistema con BullMQ se articula en torno a tres entidades principales: Queue (productor), Worker (consumidor) y QueueEvents (observador de eventos distribuidos).
[ Controlador HTTP API ]
|
v (queue.add)
+-----------------------+
| BullMQ Queue |
+-----------------------+
|
v (Script Lua Atómico)
+-------------------------------------+
| Estado Redis |
| - zset: delayed (score: timestamp) |
| - list: wait (FIFO / LIFO) |
| - hash: job metadata |
+-------------------------------------+
|
| El worker adquiere bloqueo
v (LMOVE + SET lock NX PX)
+-----------------------+
| BullMQ Worker | <----+ Renovación de Bloqueo (Watchdog)
| (concurrency: N) | | (lockRenewTime: 15s)
+-----------------------+ |
| |
¿Event Loop bloqueado? --------+
¿Worker colapsó por OOM?
|
v Bloqueo Expirado (30s)
+-------------------------------------+
| Recuperación de Tarea Stalled |
| Reubica la tarea en 'wait' o 'fail'|
+-------------------------------------+Cada tarea transita a través de una máquina de estados finitos determinista a lo largo de su ciclo de vida:
+------------+
| Delayed | (Marca de tiempo alcanzada)
+------------+
|
v
+------------+ El worker extrae la tarea
| Wait | ------------------------+
+------------+ |
^ v
| Reintento / Stalled +------------+
+------------------------- | Active |
+------------+
/ \
Resultado Éxito / \ Fallo / Intentos Agotados
v v
+-----------+ +-----------+
| Completed | | Failed |
+-----------+ +-----------+concurrency con los núcleos de CPU disponibles.El diseño de un sistema de colas robusto en Node.js se articula en tres niveles:
lockDuration), límites de concurrencia y aislamiento en sandbox.FlowProducer.import { Queue, Worker, QueueEvents, Job } from "bullmq";
import Redis from "ioredis";
export interface PaymentWebhookJobData {
transactionId: string;
amount: number;
currency: string;
source: string;
timestamp: number;
}
export interface PaymentJobResult {
processed: boolean;
ledgerRecordId: string;
}
const redisConnection = new Redis(process.env.REDIS_URL || "redis://127.0.0.1:6379", {
maxRetriesPerRequest: null, // ¡Requisito OBLIGATORIO para el funcionamiento de BullMQ!
enableReadyCheck: false,
});
export const PAYMENT_QUEUE_NAME = "payment-webhooks";
/**
* Inicialización de la Cola Productora
*/
export const paymentQueue = new Queue<PaymentWebhookJobData, PaymentJobResult>(
PAYMENT_QUEUE_NAME,
{
connection: redisConnection,
defaultJobOptions: {
attempts: 5, // Hasta 5 reintentos ante fallos imprevistos
backoff: {
type: "exponential",
delay: 2000, // Intervalo inicial de 2s, seguido de 4s, 8s, 16s...
},
removeOnComplete: {
age: 86400, // Preservar tareas completadas en Redis durante 24 horas
count: 5000, // Conservar un límite máximo de 5000 registros finalizados
},
removeOnFail: {
age: 604800, // Mantener tareas fallidas durante 7 días para inspección forense
},
},
}
);
/**
* Worker Consumidor Protegido contra Tareas Bloqueadas (Stalled Jobs)
*/
export function createPaymentWorker(): Worker<PaymentWebhookJobData, PaymentJobResult> {
const worker = new Worker<PaymentWebhookJobData, PaymentJobResult>(
PAYMENT_QUEUE_NAME,
async (job: Job<PaymentWebhookJobData, PaymentJobResult>): Promise<PaymentJobResult> => {
// Notificación de progreso para cuadros de mando en tiempo real
await job.updateProgress(10);
if (job.data.amount <= 0) {
throw new Error(`Invalid transaction amount: ${job.data.amount}`);
}
await job.updateProgress(50);
// Ejecución de la lógica transaccional de negocio
const ledgerRecordId = `led_${job.data.transactionId}`;
await job.updateProgress(100);
return {
processed: true,
ledgerRecordId,
};
},
{
connection: redisConnection,
concurrency: 10, // Hasta 10 tareas concurrentes en un único proceso worker
lockDuration: 30000, // Periodo de validez del bloqueo (30 segundos)
lockRenewTime: 10000, // Renovación del bloqueo cada 10 segundos (Watchdog)
maxStalledCount: 2, // Límite de 2 reintentos por bloqueo antes de marcar como Failed
stalledInterval: 30000, // Frecuencia de escaneo de tareas bloqueadas
}
);
// Registro de eventos críticos de telemetría
worker.on("completed", (job: Job) => {
console.info(`[Worker] Job ${job.id} completed successfully`);
});
worker.on("failed", (job: Job | undefined, err: Error) => {
console.error(`[Worker] Job ${job?.id} failed on attempt ${job?.attemptsMade}: ${err.message}`);
});
worker.on("error", (err: Error) => {
console.error("[Worker] Global worker error:", err.message);
});
worker.on("stalled", (jobId: string) => {
console.warn(`[Worker] Job ${jobId} was marked as STALLED (lock expired)`);
});
return worker;
}import { FlowProducer, FlowJob } from "bullmq";
import Redis from "ioredis";
const redis = new Redis(process.env.REDIS_URL || "redis://127.0.0.1:6379", {
maxRetriesPerRequest: null,
});
export const flowProducer = new FlowProducer({ connection: redis });
/**
* Creación de un flujo con dependencias:
* La tarea padre de agregación espera a que concluyan dos tareas hijas paralelas de extracción
*/
export async function createReportPipeline(reportId: string, companyId: string) {
const flow: FlowJob = {
name: "aggregate-final-report",
queueName: "report-aggregator",
data: { reportId, companyId },
children: [
{
name: "extract-sales-data",
queueName: "data-extraction",
data: { reportId, source: "sales" },
opts: { attempts: 3 },
},
{
name: "extract-expenses-data",
queueName: "data-extraction",
data: { reportId, source: "expenses" },
opts: { attempts: 3 },
},
],
};
const tree = await flowProducer.add(flow);
return tree.job.id;
}Un script Lua atómico genera el Hash de metadatos e inserta el ID en la lista wait o en el zset delayed.
El worker mueve la tarea de wait a active mediante LMOVE y adquiere el bloqueo durante lockDuration.
Cada 10 segundos, un temporizador interno ejecuta un script Lua PEXPIRE extendiendo el bloqueo 30 segundos.
Ante una ejecución exitosa, los datos se guardan en el Hash, la tarea pasa a completed y se libera el bloqueo.
Si el worker colapsa, el bloqueo expira. Un worker par detecta la anomalía y reencola la tarea para reintento.
version: "3.8"
services:
redis-cluster:
image: redis:7.2-alpine
command: redis-server --appendonly yes --maxmemory 2gb --maxmemory-policy noeviction
ports:
- "6379:6379"
volumes:
- redis_data:/data
restart: unless-stopped
queue-worker:
build:
context: .
dockerfile: Dockerfile.worker
deploy:
replicas: 4 # Escalado a 4 procesos independientes de Node.js
resources:
limits:
cpus: "1.5"
memory: 1024M
environment:
NODE_ENV: production
REDIS_URL: redis://redis-cluster:6379
WORKER_CONCURRENCY: 10
stop_grace_period: 45s # Margen de tiempo para permitir la finalización limpia de tareas activas
restart: on-failure
volumes:
redis_data:Ejecutar operaciones síncronas pesadas (como el parseo de un XML de 100 MB mediante JSON.parse o la computación de bcrypt.hashSync) dentro de la función de new Worker(...). El hilo principal se congela durante 15 segundos. En ese intervalo, el temporizador de renovación del bloqueo (lockRenewTime) no se puede ejecutar. En consecuencia, el lockDuration expira y otro worker marca la tarea como Stalled, reasignándola y provocando ejecuciones paralelas duplicadas. Solución: Utilizar un SandboxedProcessor (en un hilo Worker Thread o subproceso dedicado) o recurrir a APIs asíncronas basadas en streams.
Por defecto, ioredis limita los reintentos de comandos de red. Durante desconexiones transitorias de red, BullMQ depende de la acumulación de comandos en búfer hasta reanudar la conexión. Con los ajustes por defecto, ioredis arroja una excepción fatal MaxRetriesPerRequestError, interrumpiendo el proceso worker y perdiendo el contexto de las tareas. Configure siempre maxRetriesPerRequest: null.
Encolar millones de tareas sin definir removeOnComplete ni removeOnFail. Cada tarea concluida conserva su Hash de metadatos en la memoria de Redis indefinidamente. Con el tiempo, millones de claves obsoletas agotan la RAM del servidor, provocando rechazos de escritura por OOM. Configure siempre límites de retención explícitos (age: 86400, count: 5000).
Confiar en temporizadores nativos de JavaScript en servicios replicados en varios pods de Kubernetes. Cada uno de los 5 pods inicia su propio setInterval, ejecutando tareas críticas (como cobros periódicos de suscripciones) múltiples veces de forma simultánea. Las Repeatable Jobs de BullMQ sincronizan las planificaciones centralizadamente en Redis, garantizando una única ejecución por ciclo en todo el clúster.
El worker colapsa por OOM o queda atrapado en un bucle síncrono
Expira el periodo de 30 segundos de lockDuration al no emitirse latidos de renovación
Un worker par activo inspecciona active y localiza una tarea sin bloqueo válido
Se incrementa job.stalledCount; si no supera el límite, se devuelve a la lista wait
Al rebasar el umbral máximo de bloqueos, la tarea se transfiere definitivamente a failed