sickn33/bullmq-spécialiste

Que fait bullmq-spécialiste ?

Expert BullMQ pour les files d'attente de tâches basées sur Redis, le traitement en arrière-plan et l'exécution asynchrone fiable dans les applications Node.js/TypeScript.

Pourquoi cette compétence est utile

Pour les développeurs travaillant avec des systèmes de files d'attente, le traitement asynchrone ou l'architecture orientée événements, BullMQ est une bibliothèque puissante mais complexe. Cette compétence aide les agents IA et les développeurs à comprendre et implémenter correctement BullMQ, des configurations de base aux patterns avancés.

Plateformes supportées

• Node.js / TypeScript
• Requiert une instance Redis (locale, Redis Cloud, Upstash, Railway, Docker)
• Fonctionne avec des déploiements sans serveur (serverless) et des machines dédiées

Concepts clés couverts

1. Connexion Redis

import { ConnectionOptions } from 'bullmq';

const connection: ConnectionOptions = {
  host: 'localhost',
  port: 6379,
  password: 'optional'
};

2. Création d'une file d'attente

import { Queue } from 'bullmq';

const myQueue = new Queue('emails', { 
  connection,
  defaultJobOptions: {
    attempts: 3,
    backoff: { type: 'exponential', delay: 2000 },
    removeOnComplete: { age: 3600 * 24 },
    removeOnFail: { age: 3600 * 24 * 7 }
  }
});

3. Ajout de tâches (jobs)

await myQueue.add('send-welcome-email', {
  to: '[email protected]',
  template: 'welcome'
}, {
  delay: 5000,
  attempts: 2
});

4. Worker (consommateur)

import { Worker } from 'bullmq';

const worker = new Worker('emails', async job => {
  // Logique de traitement
  await sendEmail(job.data.to, job.data.template);
}, { 
  connection,
  concurrency: 10 
});

worker.on('completed', job => {
  console.log(`Tâche ${job.id} terminée`);
});

worker.on('failed', (job, err) => {
  console.error(`Tâche ${job.id} échouée:`, err);
});

5. Planificateur (Scheduler) pour tâches périodiques

import { QueueScheduler } from 'bullmq';

const scheduler = new QueueScheduler('notifications', { connection });
// Ajouter une tâche récurrente
await myQueue.add('daily-report', {}, {
  repeat: { pattern: '0 8 * * *' }
});

6. Gestion des échecs et tentatives

// Configuration des tentatives avec délai exponentiel
const job = await myQueue.add('process-payment', data, {
  attempts: 5,
  backoff: {
    type: 'exponential',
    delay: 1000
  }
});

7. Nettoyage automatique des tâches

const queue = new Queue('logs', {
  defaultJobOptions: {
    removeOnComplete: {
      count: 100,   // Garder les 100 plus récentes
      age: 3600      // OU supprimer après 1 heure
    },
    removeOnFail: {
      age: 3600 * 24 * 30 // Garder les échecs 30 jours
    }
  }
});

8. Files d'attente prioritaires

await highPriorityQueue.add('urgent-task', data, {
  priority: 1  // Priorité plus basse = exécution plus rapide
});

await lowPriorityQueue.add('background-task', data, {
  priority: 10
});

Cas d'utilisation pratiques

1. Envoi d'emails en arrière-plan : Traitement asynchrone des files d'attente d'emails avec tentatives
2. Traitement de paiements : Gestion fiable des transactions avec retry et backoff
3. Génération de rapports : Planification de rapports quotidiens ou horaires
4. Scraping web : File d'attente de tâches de scraping avec contrôle de concurrence
5. Traitement de médias : Redimensionnement d'images, transcodage vidéo

Débogage et surveillance

// Obtenir l'état de la file d'attente
const counts = await queue.getJobCounts();
// { waiting: 5, active: 2, completed: 100, failed: 3, delayed: 1 }

// Récupérer les tâches échouées
const failedJobs = await queue.getFailed();

Bonnes pratiques

• Toujours configurer removeOnComplete et removeOnFail pour éviter les fuites mémoire
• Utiliser concurrency sur le worker pour le parallélisme
• Configurer backoff pour les tâches qui peuvent échouer de manière transitoire
• Utiliser des noms de files d'attente explicites (ex: 'emails', 'notifications', 'exports')
• Gérer les erreurs proprement dans le worker (les exceptions non capturées marquent la tâche comme échouée)

Pièges courants

• Oublier de configurer la connexion Redis (défaut localhost:6379, peut ne pas fonctionner en production)
• Ne pas définir de limite de tentatives entraînant des retries infinis
• Utiliser le même worker pour plusieurs types de tâches sans déduplication (utiliser des files séparées ou des noms de tâches)
• Négliger la configuration de nettoyage (les tâches peuvent saturer la mémoire Redis)

Flux de travail avancés : Flows BullMQ

BullMQ prend en charge les workflows complexes via les Flows, permettant d'ajouter des tâches parent-enfant.

import { FlowProducer } from 'bullmq';

const flow = new FlowProducer({ connection });

const tree = await flow.add({
  name: 'finalize-order',
  data: { orderId: 123 },
  children: [
    {
      name: 'process-payment',
      data: { orderId: 123, amount: 49.99 },
      children: [
        {
          name: 'send-receipt',
          data: { orderId: 123, email: '[email protected]' }
        }
      ]
    },
    {
      name: 'inventory-check',
      data: { orderId: 123, items: ['item1', 'item2'] }
    }
  ]
});

Écouteurs d'événements globaux

const queueEvents = new QueueEvents('emails', { connection });

queueEvents.on('completed', ({ jobId }) => {
  console.log(`Tâche ${jobId} réussie`);
});

queueEvents.on('failed', ({ jobId, failedReason }) => {
  console.error(`Tâche ${jobId} échouée:`, failedReason);
});

Exemple d'utilisation de la compétence

Demandez à l'agent IA : "Configurez BullMQ avec une file d'attente prioritaire pour le traitement d'images avec un worker traitant 3 images simultanément et conservant l'historique des 100 dernières tâches terminées."

Ou : "Mettez en place un système de files d'attente avec une tâche principale qui planifie des tâches enfants pour l'envoi d'emails et la mise à jour de base de données."