bullmq / ioredis

BullMQ ejecuta colas de trabajos en Redis; ioredis es el cliente de Node que comparten tanto las colas como las capas de caché. Empareja ambos con espacios de nombres de claves claros y agrupación de conexiones.

Receta

Tarjeta de receta de referencia rápida: lista para copiar y pegar.

// src/redis.ts - fábrica de conexión única
import IORedis from "ioredis";
 
export function createRedis() {
  return new IORedis(process.env.REDIS_URL!, {
    maxRetriesPerRequest: null, // requerido para workers de BullMQ
    enableReadyCheck: false,
  });
}

// src/queues/email.queue.ts
import { Queue } from "bullmq";
import { createRedis } from "../redis.js";
 
const connection = createRedis();
 
export const emailQueue = new Queue("email", {
  connection,
  defaultJobOptions: {
    attempts: 5,
    backoff: { type: "exponential", delay: 2000 },
    removeOnComplete: 1000,
    removeOnFail: false, // mantener para inspección de DLQ
  },
});
 
await emailQueue.add("send-welcome", { userId: "abc", template: "welcome" });

// src/workers/email.worker.ts
import { Worker } from "bullmq";
import { createRedis } from "../redis.js";
 
const worker = new Worker(
  "email",
  async (job) => {
    await sendEmail(job.data);
  },
  { connection: createRedis(), concurrency: 10 }
);
 
worker.on("failed", (job, err) => {
  console.error({ jobId: job?.id, err }, "email_job_failed");
});

Cuándo usarlo:

Correo electrónico asíncrono, webhooks, generación de PDF, indexación de búsqueda
Llamadas a API de terceros con límite de velocidad a través de la concurrencia de workers
Trabajos retrasados y reemplazos de cron con trabajos repetibles de BullMQ
Caché de lecturas frecuentes con ioredis mientras los trabajos escriben en Postgres

Ejemplo de trabajo

// src/cache.ts
import type IORedis from "ioredis";
 
const CACHE_PREFIX = "cache:";
 
export async function cacheGet<T>(
  redis: IORedis,
  key: string
): Promise<T | null> {
  const raw = await redis.get(`${CACHE_PREFIX}${key}`);
  return raw ? (JSON.parse(raw) as T) : null;
}
 
export async function cacheSet(
  redis: IORedis,
  key: string,
  value: unknown,
  ttlSeconds: number
): Promise<void> {
  await redis.set(
    `${CACHE_PREFIX}${key}`,
    JSON.stringify(value),
    "EX",
    ttlSeconds
  );
}

// Manejador HTTP: patrón cache-aside
import { createRedis } from "./redis.js";
import { cacheGet, cacheSet } from "./cache.js";
 
const redis = createRedis();
 
app.get("/products/:id", async (req, reply) => {
  const { id } = req.params;
  const cached = await cacheGet<Product>(redis, `product:${id}`);
  if (cached) return cached;
 
  const product = await db.findProduct(id);
  await cacheSet(redis, `product:${id}`, product, 300);
  return product;
});

Notas de arquitectura:

BullMQ usa streams y claves de Redis bajo el prefijo bull: por defecto
La caché de la aplicación usa el prefijo cache: - sin colisiones de claves
Ejecuta los workers como despliegues separados (node dist/workers/email.js)
Los pods HTTP usan el valor por defecto de maxRetriesPerRequest para la caché; los workers usan null

Lista de verificación de operaciones

Señal	Herramienta	Acción
Retraso de cola	BullMQ `getJobCounts()`	Escalar workers
Trabajos estancados	Eventos de BullMQ	Investigar tareas de CPU largas
Memoria de Redis	`INFO memory`	Reducir `removeOnComplete`, caché TTL
Trabajos fallidos	Bull Board / Redis CLI	Reproducir o mover a DLQ

// Cola de mensajes fallidos (DLQ) después de los intentos máximos
import { Queue } from "bullmq";
 
export const dlq = new Queue("email-dlq", { connection: createRedis() });
 
worker.on("failed", async (job, err) => {
  if (job && job.attemptsMade >= (job.opts.attempts ?? 1)) {
    await dlq.add("failed", { original: job.data, error: err.message });
  }
});

Comparación

Enfoque	Elegir cuándo	Evitar cuándo
BullMQ + ioredis	Redis ya está en la pila, trabajos retrasados	Solo AWS SQS sin Redis
SQS + Lambda	Cargas de trabajo de ráfaga sin servidor	Trabajos de CPU de larga duración en proceso
pg-boss	Solo tiendas Postgres, sin Redis	Necesidades de latencia de trabajo sub-milisegundos
Listas de Redis sin procesar	Minimalismo extremo	Necesidad de reintentos, programación, UI

Preguntas frecuentes

¿El mismo Redis para caché y colas?

Sí, en pilas pequeñas/medianas con prefijos de clave y límites de memoria. A escala, separa las instancias de Redis de caché (evicción OK) de las de cola (sin evicción).

¿Cuántas conexiones por pod?

Una ioredis compartida para caché por pod HTTP; los workers necesitan conexiones dedicadas por instancia de Worker. Observa connected_clients en Redis.

¿BullMQ vs node-cron?

Usa trabajos repetibles de BullMQ cuando varias réplicas no deben disparar cron dos veces. Usa node-cron solo con elección de líder o un solo pod de programador.

Relacionado

BullMQ - patrones de cola profundos
ioredis - ajuste del cliente
Habilidad de Worker de Cola - manual del agente
Conceptos básicos de bibliotecas esenciales - emparejamiento predeterminado

Versiones de la pila: Esta página fue escrita para Node.js 24.18.0 (LTS activa), npm 10+, TypeScript 5.6+, Express 5, Fastify 5 y NestJS 11.

Receta

Tarjeta de receta de referencia rápida: lista para copiar y pegar.

// src/redis.ts - fábrica de conexión única
import IORedis from "ioredis";
 
export function createRedis() {
  return new IORedis(process.env.REDIS_URL!, {
    maxRetriesPerRequest: null, // requerido para workers de BullMQ
    enableReadyCheck: false,
  });
}

// src/queues/email.queue.ts
import { Queue } from "bullmq";
import { createRedis } from "../redis.js";
 
const connection = createRedis();
 
export const emailQueue = new Queue("email", {
  connection,
  defaultJobOptions: {
    attempts: 5,
    backoff: { type: "exponential", delay: 2000 },
    removeOnComplete: 1000,
    removeOnFail: false, // mantener para inspección de DLQ
  },
});
 
await emailQueue.add("send-welcome", { userId: "abc", template: "welcome" });

// src/workers/email.worker.ts
import { Worker } from "bullmq";
import { createRedis } from "../redis.js";
 
const worker = new Worker(
  "email",
  async (job) => {
    await sendEmail(job.data);
  },
  { connection: createRedis(), concurrency: 10 }
);
 
worker.on("failed", (job, err) => {
  console.error({ jobId: job?.id, err }, "email_job_failed");
});

Cuándo usarlo:

Correo electrónico asíncrono, webhooks, generación de PDF, indexación de búsqueda
Llamadas a API de terceros con límite de velocidad a través de la concurrencia de workers
Trabajos retrasados y reemplazos de cron con trabajos repetibles de BullMQ
Caché de lecturas frecuentes con ioredis mientras los trabajos escriben en Postgres

Ejemplo de trabajo

// src/cache.ts
import type IORedis from "ioredis";
 
const CACHE_PREFIX = "cache:";
 
export async function cacheGet<T>(
  redis: IORedis,
  key: string
): Promise<T | null> {
  const raw = await redis.get(`${CACHE_PREFIX}${key}`);
  return raw ? (JSON.parse(raw) as T) : null;
}
 
export async function cacheSet(
  redis: IORedis,
  key: string,
  value: unknown,
  ttlSeconds: number
): Promise<void> {
  await redis.set(
    `${CACHE_PREFIX}${key}`,
    JSON.stringify(value),
    "EX",
    ttlSeconds
  );
}

// Manejador HTTP: patrón cache-aside
import { createRedis } from "./redis.js";
import { cacheGet, cacheSet } from "./cache.js";
 
const redis = createRedis();
 
app.get("/products/:id", async (req, reply) => {
  const { id } = req.params;
  const cached = await cacheGet<Product>(redis, `product:${id}`);
  if (cached) return cached;
 
  const product = await db.findProduct(id);
  await cacheSet(redis, `product:${id}`, product, 300);
  return product;
});

Notas de arquitectura:

BullMQ usa streams y claves de Redis bajo el prefijo bull: por defecto
La caché de la aplicación usa el prefijo cache: - sin colisiones de claves
Ejecuta los workers como despliegues separados (node dist/workers/email.js)
Los pods HTTP usan el valor por defecto de maxRetriesPerRequest para la caché; los workers usan null

Lista de verificación de operaciones

Señal	Herramienta	Acción
Retraso de cola	BullMQ `getJobCounts()`	Escalar workers
Trabajos estancados	Eventos de BullMQ	Investigar tareas de CPU largas
Memoria de Redis	`INFO memory`	Reducir `removeOnComplete`, caché TTL
Trabajos fallidos	Bull Board / Redis CLI	Reproducir o mover a DLQ

// Cola de mensajes fallidos (DLQ) después de los intentos máximos
import { Queue } from "bullmq";
 
export const dlq = new Queue("email-dlq", { connection: createRedis() });
 
worker.on("failed", async (job, err) => {
  if (job && job.attemptsMade >= (job.opts.attempts ?? 1)) {
    await dlq.add("failed", { original: job.data, error: err.message });
  }
});

Comparación

Enfoque	Elegir cuándo	Evitar cuándo
BullMQ + ioredis	Redis ya está en la pila, trabajos retrasados	Solo AWS SQS sin Redis
SQS + Lambda	Cargas de trabajo de ráfaga sin servidor	Trabajos de CPU de larga duración en proceso
pg-boss	Solo tiendas Postgres, sin Redis	Necesidades de latencia de trabajo sub-milisegundos
Listas de Redis sin procesar	Minimalismo extremo	Necesidad de reintentos, programación, UI

Preguntas frecuentes

¿El mismo Redis para caché y colas?

Sí, en pilas pequeñas/medianas con prefijos de clave y límites de memoria. A escala, separa las instancias de Redis de caché (evicción OK) de las de cola (sin evicción).

¿Cuántas conexiones por pod?

Una ioredis compartida para caché por pod HTTP; los workers necesitan conexiones dedicadas por instancia de Worker. Observa connected_clients en Redis.

¿BullMQ vs node-cron?

Usa trabajos repetibles de BullMQ cuando varias réplicas no deben disparar cron dos veces. Usa node-cron solo con elección de líder o un solo pod de programador.

Relacionado

BullMQ - patrones de cola profundos
ioredis - ajuste del cliente
Habilidad de Worker de Cola - manual del agente
Conceptos básicos de bibliotecas esenciales - emparejamiento predeterminado

Versiones de la pila: Esta página fue escrita para Node.js 24.18.0 (LTS activa), npm 10+, TypeScript 5.6+, Express 5, Fastify 5 y NestJS 11.

bullmq / ioredis

Buscar en la documentación

bullmq / ioredis