Que fait Orchestrateur de Tâches en Arrière-Plan ?
Expert dans la conception et l'implémentation de systèmes de tâches asynchrones de qualité production qui gèrent les opérations longues sans bloquer les réponses API.
Quand l'utiliser
✅ Utilisez pour :
- Tâches longues (envoi d'emails, génération de rapports, traitement d'images)
- Opérations par lots (imports/exports en masse, migrations de données)
- Tâches planifiées (digest quotidiens, nettoyages, rapports récurrents)
- Tâches nécessitant une logique de relance (appels API externes, opérations instables)
- Traitement basé sur la priorité (utilisateurs premium d'abord, alertes critiques)
- Opérations limitées en débit (quotas API, limites de services tiers)
❌ PAS pour :
- Communication bidirectionnelle en temps réel (utilisez WebSockets)
- Latence inférieure à la seconde (utilisez le cache en mémoire)
- Simples délais (setTimeout est suffisant pour <5 secondes)
- Réponses API synchrones (gardez la logique dans le gestionnaire de requête)
Arbre de décision rapide
Est-ce que cette tâche :
├── Prend >5 secondes ? → Tâche asynchrone
├── Nécessite une relance en cas d'échec ? → Tâche asynchrone
├── S'exécute selon un planning ? → Tâche asynchrone (pattern cron)
├── Bloque l'interaction utilisateur ? → Tâche asynchrone
├── Se traite par lots ? → Tâche asynchrone
└── Renvoie immédiatement ? → Gardez synchrone
Sélection de la technologie
Node.js : BullMQ (Recommandé 2024+)
Quand l'utiliser :
- Projet TypeScript
- Redis déjà dans la stack
- Besoin de fonctionnalités avancées (limitation de débit, priorités, jobs répétables)
Pourquoi BullMQ plutôt que Bull :
- Bull (v3) → BullMQ (v4+) : Réécriture complète en TypeScript
- Meilleure gestion des connexions Redis
- Concurrence et performances améliorées
- Maintenance active (Bull est en mode maintenance)
Python : Celery
Quand l'utiliser :
- Projet Python/Django
- Besoin d'exécution distribuée de tâches
- Workflows complexes (chaînes, groupes, chords)
Alternatives :
- RQ (Redis Queue) : Plus simple, moins de fonctionnalités
- Dramatiq : Moderne, moins d'écosystème
- Huey : Léger, bon pour les petits projets
Cloud-Native : AWS SQS, Google Cloud Tasks
Quand l'utiliser :
- Architecture serverless
- Pas envie de gérer Redis/RabbitMQ
- Besoin de livraison garantie et de files de lettres mortes
Anti-patrons courants
Anti-patron 1 : Pas de file de lettres mortes
Pensée novice : "Réessayer 3 fois, puis échouer silencieusement"
Problème : Les jobs échoués disparaissent sans visibilité ni chemin de récupération.
Approche correcte :
// BullMQ avec file de lettres mortes
const queue = new Queue('file-emails', {
connection: redis,
defaultJobOptions: {
attempts: 3,
backoff: {
type: 'exponential',
delay: 2000
},
removeOnComplete: 100, // Garder les 100 derniers réussis
removeOnFail: false // Garder tous les échecs pour inspection
}
});
// Surveiller les jobs échoués
const failedJobs = await queue.getFailed();
Chronologie :
- Avant 2020 : Réessayer et oublier
- 2020+ : Files de lettres mortes standard
- 2024+ : Observabilité des échecs de jobs requise
Anti-patron 2 : Traitement synchrone des jobs
Symptôme : Le point de terminaison API attend la fin du job
Problème :
// ❌ FAUX - Bloque la réponse API
app.post('/envoyer-email', async (req, res) => {
await sendEmail(req.body.to, req.body.subject);
res.json({ success: true });
});
Pourquoi c'est faux : Timeout, mauvaise expérience utilisateur, gaspillage de ressources serveur
Approche correcte :
// ✅ CORRECT - Mettre en file et renvoyer immédiatement
app.post('/envoyer-email', async (req, res) => {
const job = await emailQueue.add('send', {
to: req.body.to,
subject: req.body.subject
});
res.json({
success: true,
jobId: job.id,
status: 'queued'
});
});
// Un worker séparé traite le job
worker.process('send', async (job) => {
await sendEmail(job.data.to, job.data.subject);
});
Anti-patron 3 : Pas d'idempotence
Problème : Le job s'exécute deux fois → doublons de frais, doubles emails
Pourquoi cela arrive :
- La connexion Redis se coupe en cours de traitement
- Le worker plante avant la fin du job
- Le timeout du job déclenche une relance alors qu'il est encore en cours
Approche correcte :
// ✅ Job idempotent avec clé de déduplication
await queue.add('paiement', {
userId: 123,
amount: 50.00
}, {
jobId: `paiement-${orderId}`, // Empêche les doublons
attempts: 3
});
// Dans le worker : Vérifier si déjà traité
worker.process('paiement', async (job) => {
const { userId, amount } = job.data;
// Vérifier l'idempotence
const existing = await db.payments.findOne({
jobId: job.id
});
if (existing) {
return existing; // Déjà traité
}
// Traiter le paiement
const result = await stripe.charges.create({...});
// Stocker l'enregistrement d'idempotence
await db.payments.create({
jobId: job.id,
result
});
return result;
});
Anti-patron 4 : Pas de limitation de débit
Problème : Surcharger les API tierces ou épuiser les quotas
Symptôme : Erreurs "Limite de débit dépassée" de Sendgrid, Stripe, etc.
Approche correcte :
// Limitation de débit BullMQ
const queue = new Queue('appels-api', {
limiter: {
max: 100, // Max 100 jobs
duration: 60000 // Par 60 secondes
}
});
// Ou : Limites de débit basées sur la priorité
await queue.add('envoyer-email', data, {
priority: user.isPremium ? 1 : 10,
rateLimiter: {
max: user.isPremium ? 1000 : 100,
duration: 3600000 // Par heure
}
});
Anti-patron 5 : Oublier le dimensionnement des workers
Problème : Un seul worker ne peut pas suivre la profondeur de la file
Symptôme : La file s'accumule, les jobs sont retardés de plusieurs heures/jours
Approche correcte :
// Scaling horizontal avec plusieurs workers
const worker = new Worker('file-emails', async (job) => {
await processEmail(job.data);
}, {
connection: redis,
concurrency: 5 // Traiter 5 jobs simultanément par worker
});
// Exécuter plusieurs processus workers (PM2, Kubernetes, etc.)
// Chaque worker traite concurrency * num_workers jobs
Monitoring :
// Configurer des alertes pour la profondeur de la file
setInterval(async () => {
const waiting = await queue.getWaitingCount();
if (waiting > 1000) {
alert('La profondeur de la file dépasse 1000, augmentez les workers !');
}
}, 60000);
Patrons d'implémentation
Patron 1 : Campagnes email
// Configuration de la file
const emailQueue = new Queue('campagne-email', { connection: redis });
// Mettre en file le lot
async function sendCampaign(userIds: number[], template: string) {
const jobs = userIds.map(userId => ({
name: 'send',
data: { userId, template },
opts: {
attempts: 3,
backoff: { type: 'exponential', delay: 5000 }
}
}));
await emailQueue.addBulk(jobs);
}
// Worker avec logique de relance
const worker = new Worker('campagne-email', async (job) => {
const { userId, template } = job.data;
const user = await db.users.findById(userId);
const email = renderTemplate(template, user);
try {
await sendgrid.send({
to: user.email,
subject: email.subject,
html: email.body
});
} catch (error) {
if (error.code === 'ECONNREFUSED') {
throw error; // Réessayer
}
// Email invalide, ne pas réessayer
console.error(`Email invalide pour l'utilisateur ${userId}`);
}
}, {
connection: redis,
concurrency: 10
});
Patron 2 : Rapports planifiés
// Rapport quotidien à 9h
await queue.add('rapport-quotidien', {
type: 'sales',
recipients: ['[email protected]']
}, {
repeat: {
pattern: '0 9 * * *', // Syntaxe cron
tz: 'America/New_York'
}
});
// Le worker génère et envoie le rapport par email
worker.process('rapport-quotidien', async (job) => {
const { type, recipients } = job.data;
const data = await generateReport(type);
const pdf = await createPDF(data);
await emailQueue.add('send', {
to: recipients,
subject: `Rapport ${type} quotidien`,
attachments: [{ filename: 'rapport.pdf', content: pdf }]
});
});
Patron 3 : Pipeline de transcodage vidéo
// Job multi-étapes avec suivi de progression
await videoQueue.add('transcode', {
videoId: 123,
formats: ['720p', '1080p', '4k']
}, {
attempts: 2,
timeout: 3600000 // Timeout de 1 heure
});
worker.process('transcode', async (job) => {
const { videoId, formats } = job.data;
for (let i = 0; i < formats.length; i++) {
const format = formats[i];
// Mettre à jour la progression
await job.updateProgress((i / formats.length) * 100);
// Transcodage
await ffmpeg.transcode(videoId, format);
}
await job.updateProgress(100);
});
// Le client interroge la progression
app.get('/videos/:id/status', async (req, res) => {
const job = await queue.getJob(req.params.jobId);
res.json({
state: await job.getState(),
progress: job.progress
});
});
Monitoring & Observabilité
Métriques essentielles
// Tableau de bord de santé de la file
async function getQueueMetrics() {
const [waiting, active, completed, failed, delayed] = await Promise.all([
queue.getWaitingCount(),
queue.getActiveCount(),
queue.getCompletedCount(),
queue.getFailedCount(),
queue.getDelayedCount()
]);
return {
waiting, // Jobs en attente de traitement
active, // Jobs en cours de traitement
completed, // Terminés avec succès
failed, // Échoués après relances
delayed, // Planifiés pour plus tard
health: waiting < 1000 && failed < 100 ? 'healthy' : 'degraded'
};
}
BullMQ Board (IU)
// Développement : Surveiller les jobs visuellement
import { createBullBoard } from '@bull-board/api';
import { BullMQAdapter } from '@bull-board/api/bullMQAdapter';
import { ExpressAdapter } from '@bull-board/express';
const serverAdapter = new ExpressAdapter();
createBullBoard({
queues: [
new BullMQAdapter(emailQueue),
new BullMQAdapter(videoQueue)
],
serverAdapter
});
app.use('/admin/queues', serverAdapter.getRouter());
// Visiter http://localhost:3000/admin/queues
Checklist de production
□ File de lettres mortes configurée
□ Stratégie de relance avec backoff exponentiel
□ Limites de timeout définies pour les jobs
□ Limitation de débit pour les API tierces
□ Clés d'idempotence pour les opérations critiques
□ Concurrence des workers ajustée (cœurs CPU * 2)
□ Scaling horizontal configuré (plusieurs workers)
□ Surveillance de la profondeur de la file avec alertes
□ Workflow d'inspection des jobs échoués
□ Les données des jobs ne contiennent pas de données personnelles dans les logs
□ Persistance Redis activée (AOF ou RDB)
□ Gestion de l'arrêt gracieux (SIGTERM)
Quand utiliser vs Éviter
| Scénario | Utiliser des tâches asynchrones ? |
|---|---|
| Envoyer un email de bienvenue lors de l'inscription | ✅ Oui - peut prendre 2-5 secondes |
| Facturer une carte de crédit | ⚠️ Peut-être - dépend de la latence du fournisseur de paiement |
| Générer un rapport PDF (30 secondes) | ✅ Oui - définitivement en arrière-plan |
| Récupérer un profil utilisateur depuis la base | ❌ Non - millisecondes, garder synchrone |
| Traiter un upload vidéo (5 minutes) | ✅ Oui - toujours en arrière-plan |
| Valider une entrée de formulaire | ❌ Non - validation synchrone |
| Tâche cron quotidienne | ✅ Oui - utiliser les jobs répétables |
| Message de chat en temps réel | ❌ Non - utiliser WebSockets |
Comparaison des technologies
| Fonctionnalité | BullMQ | Celery | AWS SQS |
|---|---|---|---|
| Langage | Node.js | Python | Tous (API HTTP) |
| Backend | Redis | Redis/RabbitMQ/SQS | Géré |
| Priorités | ✅ | ✅ | ✅ |
| Limitation de débit | ✅ | ❌ | ✅ (via attributs) |
| Répétition/Cron | ✅ | ✅ (celery-beat) | ❌ (utiliser EventBridge) |
| IU de tableau de bord | Bull Board | Flower | CloudWatch |
| Workflows | ❌ | ✅ (chaînes, groupes) | ❌ |
| Courbe d'apprentissage | Moyenne | Moyenne | Faible |
| Coût | Hébergement Redis | Hébergement Redis | 0,40 $/million de requêtes |
Références
/references/bullmq-patterns.md- Patrons et exemples avancés BullMQ/references/celery-workflows.md- Chaînes, groupes et chords Celery/references/job-observability.md- Monitoring, alerting et débogage
Scripts
scripts/setup_bullmq.sh- Initialiser BullMQ avec Redisscripts/queue_health_check.ts- Tableau de bord des métriques de la filescripts/retry_failed_jobs.ts- Relance en masse des jobs échoués
Ce skill guide : Implémentation de tâches asynchrones | Architecture de files d'attente | Stratégies de relance | Dimensionnement des workers | Observabilité des jobs