BullMQ Muster

Kurzanleitung: Verwenden Sie BullMQ (v5.x) für die Verarbeitung von Hintergrundjobs, Aufgabenplanung und Workflow-Orchestrierung auf Basis von Redis. Kernklassen: Queue (fügt Jobs hinzu), Worker (verarbeitet Jobs), QueueEvents (globaler Ereignis-Listener), FlowProducer (Eltern-Kind-Job-Bäume). Übergeben Sie immer ein connection-Objekt an jeden Konstruktor (erforderlich in v5). Setzen Sie maxRetriesPerRequest: null auf ioredis-Verbindungen für Worker. Verwenden Sie upsertJobScheduler für wiederholbare/Cron-Jobs (ersetzt die veraltete wiederholbare API). QueueScheduler wurde in v4 entfernt – seine Aufgaben werden jetzt automatisch von Workern übernommen.

<critical_requirements>

KRITISCH: Vor der Verwendung dieser Fähigkeit

Der gesamte Code muss den Projektkonventionen in CLAUDE.md folgen (kebab-case, benannte Exporte, Import-Reihenfolge, import type, benannte Konstanten)

(Sie MÜSSEN ein connection-Objekt an jeden Queue-, Worker-, QueueEvents- und FlowProducer-Konstruktor übergeben – BullMQ v5 wirft einen Fehler, wenn die Verbindung fehlt)

(Sie MÜSSEN maxRetriesPerRequest: null für ioredis-Verbindungen setzen, die von Workern verwendet werden – BullMQ erfordert unendliche Wiederholungen und wirft einen Fehler ohne diese Einstellung)

(Sie MÜSSEN await worker.close() bei SIGTERM/SIGINT für ein Graceful Shutdown aufrufen – ohne dies werden laufende Jobs zu „stalled“ Jobs)

(Sie MÜSSEN upsertJobScheduler für wiederholbare/Cron-Jobs verwenden – die alte repeat-Option bei queue.add ist seit v5.16.0 veraltet)

</critical_requirements>

Beispiele

Kernmuster – Queue-Setup, Worker-Verarbeitung, Job-Optionen, Connection-Factory, Graceful Shutdown, typisierte Jobs
Fortgeschrittene Muster – FlowProducer, Ratenbegrenzung, Job-Planung, QueueEvents, Parallelverarbeitung, Sandbox-Prozessoren

Zusätzliche Ressourcen:

reference.md – Entscheidungsrahmen, Job-Optionen-Referenz, Anti-Patterns, Produktions-Checkliste

Auto-Erkennung: BullMQ, bullmq, Queue, Worker, QueueEvents, FlowProducer, Job Queue, Background Job, Worker Process, Job Scheduler, upsertJobScheduler, Rate Limiter, Job Priority, Job Delay, Sandboxed Processor, Repeatable Job, Cron Job, Flow Producer, Parent Child Jobs

Wann verwenden?

Hintergrundverarbeitung (E-Mail-Versand, Bildverarbeitung, PDF-Generierung)
Geplante/Cron-Jobs (nächtliche Berichte, regelmäßige Bereinigung)
Workflow-Orchestrierung mit Eltern-Kind-Job-Abhängigkeiten (FlowProducer)
Ratenbegrenzte API-Nutzung (Drosselung ausgehender Anfragen)
Prioritätsbasierte Jobverarbeitung (dringende Jobs vor Massenvorgängen)
Verteilung CPU-intensiver Arbeit auf mehrere Worker oder Maschinen

Abgedeckte Schlüsselmuster:

Queue- und Worker-Setup mit typisierten Jobdaten und Rückgabewerten
Connection-Factory mit maxRetriesPerRequest: null für Worker
Job-Optionen: Verzögerung, Priorität, Versuche, Backoff, removeOnComplete/Fail
Graceful Shutdown mit worker.close() bei Prozesssignalen
FlowProducer für Eltern-Kind-Job-Bäume mit Abhängigkeitsverfolgung
Job Scheduler für wiederholbare/Cron-Jobs (upsertJobScheduler)
Ratenbegrenzung (globaler Limiter und manueller Worker.RateLimitError)
Parallelitätskontrolle (lokal pro Worker und global)
QueueEvents für globale Ereignisüberwachung über alle Worker hinweg
Sandbox-Prozessoren für CPU-intensive Arbeit

Wann NICHT verwenden?

Einfache In-Prozess-Timer oder setTimeout (keine Persistenz erforderlich)
Echtzeit-Pub/Sub-Messaging ohne Persistenz (verwenden Sie Ihre Pub/Sub-Lösung)
Daten, die synchron innerhalb eines Request-Response-Zyklus verarbeitet werden müssen
Warteschlangen, die keine Persistenz, Wiederholungen oder Planung benötigen

Philosophie

BullMQ ist eine Redis-gestützte Job-Warteschlange für Node.js, die zuverlässige Hintergrundverarbeitung mit Mindestens-Einmal-Zustellungsgarantie bietet. Das Kernprinzip: Trennung von Job-Produktion und Job-Konsumption, damit Ihre Anwendung reaktionsfähig bleibt, während die Arbeit asynchron erledigt wird.

Kernprinzipien:

Produzenten und Konsumenten sind entkoppelt – Jeder Prozess kann Jobs zu einer Warteschlange hinzufügen; jeder Worker kann sie verarbeiten. Dies ermöglicht horizontale Skalierung durch Hinzufügen weiterer Worker.
Jobs sind persistent – Jobs überleben Prozessneustarts, da sie in Redis leben. Ein abgestürzter Worker wird von anderen Workern aufgegriffen (oder vom selben Worker nach einem Neustart).
Mindestens-Einmal-Zustellung – BullMQ garantiert, dass jeder Job mindestens einmal verarbeitet wird. Verwenden Sie idempotente Prozessoren, um den (seltenen) Fall einer doppelten Verarbeitung nach einem „Stall“ zu behandeln.
Graceful Fail mit Wiederholungen – Konfigurieren Sie attempts- und backoff-Strategien, damit vorübergehende Fehler automatisch behoben werden. Dauerhaft fehlgeschlagene Jobs werden zur Überprüfung in den fehlgeschlagenen Satz verschoben.
Jede Queue/Worker/QueueEvents benötigt eine eigene Verbindung – BullMQ verwaltet den Verbindungsstatus intern. Teilen Sie niemals eine einzelne ioredis-Instanz über mehrere BullMQ-Klassen hinweg (außer Queues, die nur als Produzenten fungieren).

Kernmuster

Muster 1: Connection-Factory

BullMQ v5 erfordert eine explizite Redis-Verbindung für jeden Konstruktor. Worker benötigen maxRetriesPerRequest: null, damit ioredis unbegrenzt wiederholt, anstatt aufzugeben.

import Redis from "ioredis";

function createBullMQConnection(): Redis {
  const url = process.env.REDIS_URL;
  if (!url) throw new Error("REDIS_URL is required");
  return new Redis(url, { maxRetriesPerRequest: null });
}
export { createBullMQConnection };

Warum gut? maxRetriesPerRequest: null erfüllt BullMQs Anforderung, die Factory gewährleistet konsistente Konfiguration, die Umgebungsvariable hält Anmeldeinformationen aus dem Code heraus

// Schlecht – fehlendes maxRetriesPerRequest
const redis = new Redis("redis://localhost:6379");
const worker = new Worker("emails", processor, { connection: redis });
// BullMQ wirft: "maxRetriesPerRequest must be null"

Warum schlecht? BullMQ erfordert unendliche Wiederholungen für Worker-Verbindungen und wird ohne null beim Start einen Fehler werfen

Siehe examples/core.md für die vollständige Connection-Factory mit Trennung von Produzent und Konsument.

Muster 2: Queue- und Worker-Setup

Queue fügt Jobs hinzu; Worker verarbeitet sie. Beide akzeptieren TypeScript-Generics für typsichere Jobdaten und Rückgabewerte.

import { Queue, Worker, type Job } from "bullmq";

interface EmailJobData {
  to: string;
  subject: string;
  body: string;
}

const QUEUE_NAME = "emails";

const emailQueue = new Queue<EmailJobData>(QUEUE_NAME, {
  connection: createBullMQConnection(),
});

const emailWorker = new Worker<EmailJobData>(
  QUEUE_NAME,
  async (job: Job<EmailJobData>) => {
    await sendEmail(job.data.to, job.data.subject, job.data.body);
  },
  { connection: createBullMQConnection() },
);

Warum gut? Generics erzwingen Typsicherheit bei job.data, getrennte Verbindungen für Queue und Worker, benannte Konstante für den Warteschlangennamen

Siehe examples/core.md für das vollständige Setup mit Ereignissen, Fehlerbehandlung und typisierten Rückgabewerten.

Muster 3: Job-Optionen

Steuern Sie das Job-Verhalten mit Optionen: Verzögerung, Priorität, Wiederholungen, Backoff und automatische Entfernung.

const MAX_ATTEMPTS = 5;
const BACKOFF_DELAY_MS = 1000;
const DELAY_MS = 60_000;
const KEEP_COMPLETED_COUNT = 100;

await emailQueue.add(
  "send-welcome",
  { to: "[email protected]", subject: "Welcome", body: "..." },
  {
    delay: DELAY_MS,
    priority: 1, // 1 = höchste, höhere Zahlen = niedrigere Priorität
    attempts: MAX_ATTEMPTS,
    backoff: { type: "exponential", delay: BACKOFF_DELAY_MS },
    removeOnComplete: { count: KEEP_COMPLETED_COUNT },
    removeOnFail: false,
  },
);

Warum gut? Benannte Konstanten für alle numerischen Werte, exponentieller Backoff für vorübergehende Fehler, removeOnComplete mit count verhindert unbegrenztes Redis-Wachstum, removeOnFail: false bewahrt fehlgeschlagene Jobs zum Debuggen

Siehe examples/core.md für defaultJobOptions, addBulk und LIFO-Muster.

Muster 4: Graceful Shutdown

Rufen Sie worker.close() bei Prozesssignalen auf, um laufende Jobs vor dem Beenden abzuschließen. Ohne dies werden Jobs zu „stalled“ und von anderen Workern erneut verarbeitet.

async function shutdown(workers: Worker[]): Promise<void> {
  await Promise.all(workers.map((w) => w.close()));
  process.exit(0);
}

process.on("SIGTERM", () => shutdown([emailWorker]));
process.on("SIGINT", () => shutdown([emailWorker]));

Warum gut? close() stoppt die Annahme neuer Jobs und wartet auf den Abschluss laufender Jobs, Promise.all behandelt mehrere Worker, Signal-Handler decken sowohl SIGTERM als auch SIGINT ab

Achtung: worker.close() hat kein integriertes Timeout. Wenn ein Prozessor hängt, hängt auch das Herunterfahren. Wickeln Sie es bei Bedarf mit Ihrem eigenen Timeout ein.

Siehe examples/core.md für Herunterfahren mit Timeout-Muster.

Muster 5: FlowProducer (Eltern-Kind-Jobs)

FlowProducer erstellt Job-Bäume, bei denen Eltern-Jobs darauf warten, dass alle Kinder abgeschlossen sind. Der gesamte Baum wird atomar hinzugefügt.

import { FlowProducer } from "bullmq";

const flowProducer = new FlowProducer({ connection: createBullMQConnection() });

await flowProducer.add({
  name: "publish-video",
  queueName: "videos",
  children: [
    { name: "extract-audio", queueName: "media", data: { videoId: "abc" } },
    {
      name: "generate-thumbnail",
      queueName: "media",
      data: { videoId: "abc" },
    },
    {
      name: "transcode",
      queueName: "media",
      data: { videoId: "abc", format: "mp4" },
    },
  ],
});

Warum gut? Eltern warten, bis alle Kinder abgeschlossen sind, atomare Hinzufügung (alles oder nichts), Kinder können in verschiedenen Warteschlangen sein, Eltern-Prozessor kann über job.getChildrenValues() auf Kinderergebnisse zugreifen

Siehe examples/advanced.md für den Zugriff auf Kinderwerte und verschachtelte Flows.

Muster 6: Job Scheduler (Wiederholbare/Cron-Jobs)

Verwenden Sie upsertJobScheduler (v5.16.0+) für wiederholbare Jobs. Ersetzt die veraltete repeat-Option.

const REPORT_INTERVAL_MS = 3_600_000; // 1 Stunde

// Festes Intervall
await emailQueue.upsertJobScheduler(
  "hourly-digest",
  { every: REPORT_INTERVAL_MS },
  {
    name: "send-digest",
    data: { type: "hourly" },
  },
);

// Cron-Ausdruck – täglich um 3:15 Uhr
await emailQueue.upsertJobScheduler(
  "nightly-cleanup",
  { pattern: "0 15 3 * * *" },
  {
    name: "cleanup",
    data: { type: "nightly" },
    opts: { removeOnComplete: true },
  },
);

Warum gut? upsertJobScheduler ist idempotent (sicher bei jedem Start aufzurufen), die Vorlage definiert Job-Name/Daten/Optionen, die von jedem erzeugten Job geerbt werden, Cron-Syntax über pattern

Achtung: every-Intervalle richten sich nach der Uhr, nicht nach dem Zeitpunkt, an dem Sie upsertJobScheduler aufgerufen haben. Ein 2000ms-Intervall feuert bei 0s, 2s, 4s usw.

Siehe examples/advanced.md zum Entfernen von Schedulern und Auflisten aktiver Scheduler.

Muster 7: Ratenbegrenzung

Begrenzen Sie, wie viele Jobs ein Worker pro Zeitfenster verarbeitet. Der Limiter ist global über alle Worker in derselben Warteschlange hinweg.

const RATE_LIMIT_MAX = 10;
const RATE_LIMIT_DURATION_MS = 1000;

const worker = new Worker("api-calls", processor, {
  connection: createBullMQConnection(),
  limiter: { max: RATE_LIMIT_MAX, duration: RATE_LIMIT_DURATION_MS },
});

Warum gut? Globaler Ratenbegrenzer (10 Worker mit dieser Konfiguration verarbeiten immer noch maximal 10 Jobs/Sekunde insgesamt), benannte Konstanten für Limits

Für dynamische Ratenbegrenzung basierend auf externen API-Antworten verwenden Sie worker.rateLimit(duration) + throw Worker.RateLimitError(). Siehe examples/advanced.md.

Muster 8: QueueEvents für globale Überwachung

QueueEvents verwendet Redis Streams (nicht Pub/Sub), sodass Ereignisse auch bei Wiederverbindungen zuverlässig sind. Erfordert eine eigene dedizierte Verbindung.

import { QueueEvents } from "bullmq";

const queueEvents = new QueueEvents(QUEUE_NAME, {
  connection: createBullMQConnection(),
});

queueEvents.on("completed", ({ jobId, returnvalue }) => {
  console.log(`Job ${jobId} completed with: ${returnvalue}`);
});

queueEvents.on("failed", ({ jobId, failedReason }) => {
  console.error(`Job ${jobId} failed: ${failedReason}`);
});

Warum gut? Überwacht alle Worker einer Warteschlange von einem einzigen Listener aus, zuverlässige Zustellung über Redis Streams, separate Verbindung wie erforderlich

Siehe examples/advanced.md für die Fortschrittsverfolgung und das Warten auf den Abschluss bestimmter Jobs.

Leistungsoptimierung

Parallelität – Setzen Sie concurrency in den Worker-Optionen, um mehrere Jobs parallel innerhalb einer einzelnen Worker-Instanz zu verarbeiten: { concurrency: 10 }. Nur effektiv für I/O-gebundene Arbeit. CPU-gebundene Arbeit blockiert die Ereignisschleife und verursacht „stalled“ Jobs – verwenden Sie stattdessen Sandbox-Prozessoren.
Sandbox-Prozessoren – Übergeben Sie einen Dateipfad anstelle einer Funktion an den Worker-Konstruktor, um Prozessoren in einem separaten Prozess auszuführen (oder Worker-Thread mit useWorkerThreads: true). Verhindert, dass CPU-intensive Arbeit die Sperrverlängerung blockiert.
addBulk – Verwenden Sie queue.addBulk([...]), um viele Jobs in einem einzigen Redis-Roundtrip hinzuzufügen, anstatt queue.add() in einer Schleife aufzurufen.
removeOnComplete/Fail – Konfigurieren Sie die automatische Entfernung, um unbegrenztes Redis-Wachstum zu verhindern. Verwenden Sie { count: N }, um die letzten N Jobs zum Debuggen aufzubewahren.
Separate Verbindungen – Produzenten (Queue) können eine Verbindung teilen. Worker und QueueEvents benötigen jeweils eine eigene Verbindung aufgrund blockierender Redis-Befehle.

<decision_framework>

Entscheidungsrahmen

Wann BullMQ verwenden?

Benötigen Sie Hintergrund-Job-Verarbeitung?
|-- NEIN -> BullMQ nicht verwenden
+-- JA -> Benötigen Sie Persistenz, Wiederholungen oder Planung?
    |-- NEIN -> Einfache In-Prozess-Warteschlange oder setTimeout könnte ausreichen
    +-- JA -> Benötigen Sie Eltern-Kind-Job-Abhängigkeiten?
        |-- JA -> BullMQ mit FlowProducer
        +-- NEIN -> Benötigen Sie Ratenbegrenzung oder Priorität?
            |-- JA -> BullMQ mit Limiter/Prioritätsoptionen
            +-- NEIN -> BullMQ mit einfacher Queue + Worker

Welches Job-Muster?

Welche Art von Job-Planung benötigen Sie?
|-- Einmaliger verzögerter Job -> queue.add() mit delay-Option

…

[Gekürzt für Übersetzung]

agents-inc/skills

Was ist skills?

In Ihrer bevorzugten KI fragen

Dokumentation