Modèles BullMQ

Guide rapide : Utilisez BullMQ (v5.x) pour le traitement de jobs en arrière-plan, la planification de tâches et l'orchestration de workflows basée sur Redis. Classes principales : Queue (ajoute des jobs), Worker (traite des jobs), QueueEvents (écouteur d'événements global), FlowProducer (arbres de jobs parent-enfant). Passez toujours un objet connection à chaque constructeur (obligatoire dans v5). Définissez maxRetriesPerRequest: null sur les connexions ioredis pour les Workers. Utilisez upsertJobScheduler pour les jobs répétables/cron (remplace l'API répétable dépréciée). QueueScheduler a été supprimé dans v4 — ses responsabilités sont maintenant gérées automatiquement par les Workers.

<critical_requirements>

CRITIQUE : Avant d'utiliser cette compétence

Tout le code doit suivre les conventions du projet dans CLAUDE.md (kebab-case, exports nommés, ordre des importations, import type, constantes nommées)

(Vous DEVEZ passer un objet connection à chaque constructeur Queue, Worker, QueueEvents et FlowProducer — BullMQ v5 génère une erreur si la connexion est manquante)

(Vous DEVEZ définir maxRetriesPerRequest: null sur les connexions ioredis utilisées par les Workers — BullMQ nécessite des tentatives infinies et génère une erreur sans ce paramètre)

(Vous DEVEZ appeler await worker.close() sur SIGTERM/SIGINT pour un arrêt gracieux — sans cela, les jobs en cours deviennent bloqués)

(Vous DEVEZ utiliser upsertJobScheduler pour les jobs répétables/cron — l'ancienne option repeat sur queue.add est dépréciée depuis v5.16.0)

</critical_requirements>

Exemples

Modèles de base — Configuration de file d'attente, traitement Worker, options de jobs, fabrique de connexion, arrêt gracieux, jobs typés
Modèles avancés — FlowProducer, limitation de débit, planification de jobs, QueueEvents, concurrence, processeurs sandboxés

Ressources supplémentaires :

reference.md — Cadres de décision, référence des options de jobs, anti-modèles, liste de vérification pour la production

Détection automatique : BullMQ, bullmq, Queue, Worker, QueueEvents, FlowProducer, file d'attente de jobs, job en arrière-plan, processus worker, planificateur de jobs, upsertJobScheduler, limiteur de débit, priorité de job, délai de job, processeur sandboxé, job répétable, job cron, producteur de flux, jobs parent-enfant

Quand utiliser :

Traitement en arrière-plan (envoi d'e-mails, traitement d'images, génération de PDF)
Jobs planifiés/cron (rapports nocturnes, nettoyage périodique)
Orchestration de workflows avec dépendances parent-enfant (FlowProducer)
Consommation d'API limitée en débit (limitation des requêtes sortantes)
Traitement de jobs basé sur la priorité (jobs urgents avant les opérations groupées)
Distribution de travail intensif en CPU sur plusieurs workers ou machines

Modèles clés couverts :

Configuration de Queue et Worker avec types de données de job et valeurs de retour typées
Fabrique de connexion avec maxRetriesPerRequest: null pour les Workers
Options de jobs : délai, priorité, tentatives, backoff, removeOnComplete/Fail
Arrêt gracieux avec worker.close() sur les signaux de processus
FlowProducer pour les arbres de jobs parent-enfant avec suivi des dépendances
Planificateurs de jobs pour les jobs répétables/cron (upsertJobScheduler)
Limitation de débit (limiteur global et Worker.RateLimitError manuel)
Contrôle de la concurrence (local par worker et global)
QueueEvents pour la surveillance globale des événements sur tous les workers
Processeurs sandboxés pour le travail intensif en CPU

Quand NE PAS utiliser :

Minuteurs simples in-process ou setTimeout (aucune persistance nécessaire)
Messagerie pub/sub en temps réel sans persistance (utilisez votre solution pub/sub)
Données qui doivent être traitées de manière synchrone dans un cycle requête-réponse
Files d'attente qui n'ont pas besoin de persistance, de tentatives ou de planification

Philosophie

BullMQ est une file d'attente de jobs basée sur Redis pour Node.js qui fournit un traitement fiable en arrière-plan avec des garanties de livraison au moins une fois. Le principe fondamental : séparer la production de jobs de leur consommation afin que votre application reste réactive pendant que le travail s'effectue de manière asynchrone.

Principes fondamentaux :

Les producteurs et les consommateurs sont découplés — Tout processus peut ajouter des jobs à une file d'attente ; tout Worker peut les traiter. Cela permet une mise à l'échelle horizontale en ajoutant plus de Workers.
Les jobs sont persistants — Les jobs survivent aux redémarrages de processus car ils vivent dans Redis. Les jobs d'un Worker qui a planté sont repris par d'autres Workers (ou le même Worker après redémarrage).
Livraison au moins une fois — BullMQ garantit que chaque job est traité au moins une fois. Utilisez des processeurs idempotents pour gérer le cas (rare) de traitement en double après un blocage.
Échouez gracieusement avec des tentatives — Configurez des stratégies attempts et backoff pour que les échecs transitoires se résolvent automatiquement. Les jobs qui échouent définitivement passent dans l'ensemble des échecs pour inspection.
Chaque Queue/Worker/QueueEvents a besoin de sa propre connexion — BullMQ gère l'état de la connexion en interne. Ne partagez jamais une seule instance ioredis entre plusieurs classes BullMQ (sauf les Queues agissant uniquement comme producteurs).

Modèles de base

Modèle 1 : Fabrique de connexion

BullMQ v5 nécessite une connexion Redis explicite sur chaque constructeur. Les Workers ont besoin de maxRetriesPerRequest: null pour qu'ioredis réessaye indéfiniment au lieu d'abandonner.

import Redis from "ioredis";

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

Pourquoi c'est bon : maxRetriesPerRequest: null satisfait à l'exigence de BullMQ, la fabrique assure une configuration cohérente, la variable d'environnement garde les identifiants hors du code

// Mauvais -- maxRetriesPerRequest manquant
const redis = new Redis("redis://localhost:6379");
const worker = new Worker("emails", processor, { connection: redis });
// BullMQ génère : "maxRetriesPerRequest doit être null"

Pourquoi c'est mauvais : BullMQ nécessite des tentatives infinies sur les connexions Worker et générera une erreur au démarrage sans null

Voir examples/core.md pour la fabrique de connexion complète avec séparation producteur vs consommateur.

Modèle 2 : Configuration de Queue et Worker

Queue ajoute des jobs ; Worker les traite. Les deux acceptent des génériques TypeScript pour des types de données de job et des valeurs de retour sécurisés.

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

Pourquoi c'est bon : Les génériques imposent la sécurité de type sur job.data, connexions séparées pour Queue et Worker, constante nommée pour le nom de la file d'attente

Voir examples/core.md pour une configuration complète avec événements, gestion des erreurs et valeurs de retour typées.

Modèle 3 : Options de jobs

Contrôlez le comportement des jobs avec des options : délai, priorité, tentatives, backoff et suppression automatique.

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: "Bienvenue", body: "..." },
  {
    delay: DELAY_MS,
    priority: 1, // 1 = le plus élevé, des nombres plus élevés = priorité plus faible
    attempts: MAX_ATTEMPTS,
    backoff: { type: "exponential", delay: BACKOFF_DELAY_MS },
    removeOnComplete: { count: KEEP_COMPLETED_COUNT },
    removeOnFail: false,
  },
);

Pourquoi c'est bon : Constantes nommées pour toutes les valeurs numériques, backoff exponentiel pour les échecs transitoires, removeOnComplete avec count empêche une croissance illimitée de la mémoire Redis, removeOnFail: false préserve les jobs échoués pour le débogage

Voir examples/core.md pour defaultJobOptions, addBulk et les modèles LIFO.

Modèle 4 : Arrêt gracieux

Appelez worker.close() sur les signaux de processus pour terminer les jobs en cours avant de quitter. Sans cela, les jobs deviennent bloqués et sont retraités par d'autres Workers.

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

Pourquoi c'est bon : close() arrête d'accepter de nouveaux jobs et attend la fin des jobs en cours, Promise.all gère plusieurs workers, les gestionnaires de signaux couvrent à la fois SIGTERM et SIGINT

Piège : worker.close() n'a pas de délai d'attente intégré. Si un processeur se bloque, l'arrêt se bloque. Enveloppez-le avec votre propre délai d'attente si nécessaire.

Voir examples/core.md pour l'arrêt avec modèle de délai d'attente.

Modèle 5 : FlowProducer (jobs parent-enfant)

FlowProducer crée des arbres de jobs où les jobs parents attendent que tous les enfants soient terminés. L'arbre entier est ajouté de manière atomique.

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" },
    },
  ],
});

Pourquoi c'est bon : Le parent attend que tous les enfants soient terminés, ajout atomique (tout ou rien), les enfants peuvent être dans différentes files d'attente, le processeur parent peut accéder aux résultats des enfants via job.getChildrenValues()

Voir examples/advanced.md pour accéder aux valeurs des enfants et les flux imbriqués.

Modèle 6 : Planificateurs de jobs (jobs répétables/cron)

Utilisez upsertJobScheduler (v5.16.0+) pour les jobs répétables. Remplace l'option repeat dépréciée.

const REPORT_INTERVAL_MS = 3_600_000; // 1 heure

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

// Expression cron -- quotidien à 3h15 du matin
await emailQueue.upsertJobScheduler(
  "nightly-cleanup",
  { pattern: "0 15 3 * * *" },
  {
    name: "cleanup",
    data: { type: "nightly" },
    opts: { removeOnComplete: true },
  },
);

Pourquoi c'est bon : upsertJobScheduler est idempotent (sûr à appeler à chaque démarrage), le modèle définit le nom/les données/les options du job hérités par chaque job produit, la syntaxe cron via pattern

Piège : Les intervalles every s'alignent sur l'horloge, pas sur le moment où vous avez appelé upsertJobScheduler. Un intervalle de 2000 ms se déclenche à 0s, 2s, 4s, etc.

Voir examples/advanced.md pour supprimer des planificateurs et lister les planificateurs actifs.

Modèle 7 : Limitation de débit

Limitez le nombre de jobs qu'un Worker traite par fenêtre de temps. Le limiteur est global sur tous les Workers de la même file d'attente.

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

Pourquoi c'est bon : Limite de débit globale (10 workers avec cette configuration traitent toujours un maximum de 10 jobs/seconde au total), constantes nommées pour les limites

Pour la limitation de débit dynamique basée sur les réponses d'API externes, utilisez worker.rateLimit(duration) + throw Worker.RateLimitError(). Voir examples/advanced.md.

Modèle 8 : QueueEvents pour la surveillance globale

QueueEvents utilise Redis Streams (pas Pub/Sub) donc les événements sont fiables même lors des reconnexions. Nécessite sa propre connexion dédiée.

import { QueueEvents } from "bullmq";

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

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

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

Pourquoi c'est bon : Surveille tous les Workers sur une file d'attente depuis un seul écouteur, livraison fiable via Redis Streams, connexion séparée comme requis

Voir examples/advanced.md pour le suivi de la progression et l'attente de l'achèvement d'un job spécifique.

Optimisation des performances

Concurrence — Définissez concurrency sur les options Worker pour traiter plusieurs jobs en parallèle dans une seule instance Worker : { concurrency: 10 }. Efficace uniquement pour le travail lié aux E/S. Le travail lié au CPU bloque la boucle d'événements et provoque des jobs bloqués — utilisez plutôt des processeurs sandboxés.
Processeurs sandboxés — Passez un chemin de fichier au lieu d'une fonction au constructeur Worker pour exécuter les processeurs dans un processus séparé (ou un thread worker avec useWorkerThreads: true). Empêche le travail intensif en CPU de bloquer le renouvellement de verrouillage.
addBulk — Utilisez queue.addBulk([...]) pour ajouter plusieurs jobs en un seul aller-retour Redis au lieu d'appeler queue.add() en boucle.
removeOnComplete/Fail — Configurez la suppression automatique pour empêcher une croissance illimitée de la mémoire Redis. Utilisez { count: N } pour conserver les N derniers jobs pour le débogage.
Connexions séparées — Les producteurs (Queue) peuvent partager une connexion. Les Workers et QueueEvents ont chacun besoin de leur propre connexion en raison des commandes Redis bloquantes.

<decision_framework>

Cadre de décision

Quand utiliser BullMQ

Avez-vous besoin d'un traitement de jobs en arrière-plan ?
|-- NON -> N'utilisez pas BullMQ
+-- OUI -> Avez-vous besoin de persistance, de tentatives ou de planification ?
    |-- NON -> Une file d'attente simple in-process ou setTimeout peut suffire
    +-- OUI -> Avez-vous besoin de dépendances parent-enfant ?
        |-- OUI -> BullMQ avec FlowProducer
        +-- NON -> Avez-vous besoin de limitation de débit ou de priorité ?
            |-- OUI -> BullMQ avec options de limiteur/priorité
            +-- NON -> BullMQ avec Queue + Worker de base

Quel modèle de job ?

De quel type de planification de jobs avez-vous besoin ?
|-- Job différé unique -> queue.add() avec option delay

…

[tronqué pour la traduction]

</decision_framework>

agents-inc/skills

Qu'est-ce que skills ?

Demander à votre IA préférée

Documentation