Was macht Hintergrundjob-Orchestrator?
Experte für die Entwicklung und Implementierung von produktionsreifen Hintergrundjob-Systemen, die langlaufende Aufgaben ausführen, ohne API-Antworten zu blockieren.
Verwendungshinweise
✅ Verwenden für:
- Langlaufende Aufgaben (E-Mail-Versand, Berichtserstellung, Bildverarbeitung)
- Batch-Operationen (Massenimporte, -exporte, Datenmigrationen)
- Zeitgesteuerte Aufgaben (tägliche Zusammenfassungen, Aufräumjobs, wiederkehrende Berichte)
- Aufgaben mit Wiederholungslogik (externe API-Aufrufe, instabile Operationen)
- Prioritätsbasierte Verarbeitung (Premium-Benutzer zuerst, kritische Benachrichtigungen)
- Ratenbegrenzte Operationen (API-Kontingente, Beschränkungen von Drittanbietern)
❌ NICHT verwenden für:
- Echtzeit-Bidirektionale Kommunikation (WebSockets verwenden)
- Anforderungen mit Sub-Sekunden-Latenz (In-Memory-Caching verwenden)
- Einfache Verzögerungen (setTimeout ist für <5 Sekunden geeignet)
- Synchrone API-Antworten (Logik im Request-Handler belassen)
Schnellentscheidungsbaum
Erledigt diese Aufgabe:
├── Dauer >5 Sekunden? → Hintergrundjob
├── Wiederholung bei Fehlschlag erforderlich? → Hintergrundjob
├── Läuft nach Zeitplan? → Hintergrundjob (Cron-Muster)
├── Blockiert Benutzerinteraktion? → Hintergrundjob
├── Verarbeitung in Batches? → Hintergrundjob
└── Sofortige Rückgabe? → Synchron bleiben
Technologieauswahl
Node.js: BullMQ (Empfohlen ab 2024)
Wann verwenden:
- TypeScript-Projekt
- Redis bereits im Stack vorhanden
- Erweiterte Funktionen erforderlich (Rate Limiting, Prioritäten, wiederholbare Jobs)
Warum BullMQ statt Bull:
- Bull (v3) → BullMQ (v4+): Vollständige Neuschreibung in TypeScript
- Bessere Redis-Verbindungsverwaltung
- Verbesserte Parallelität und Leistung
- Aktive Wartung (Bull ist im Wartungsmodus)
Python: Celery
Wann verwenden:
- Python/Django-Projekt
- Verteilte Aufgabenausführung erforderlich
- Komplexe Workflows (Chains, Groups, Chords)
Alternativen:
- RQ (Redis Queue): Einfacher, weniger Funktionen
- Dramatiq: Moderner, weniger Ökosystem
- Huey: Leichtgewichtig, gut für kleine Projekte
Cloud-Nativ: AWS SQS, Google Cloud Tasks
Wann verwenden:
- Serverlose Architektur
- Kein Wunsch, Redis/RabbitMQ zu verwalten
- Garantierte Zustellung und Dead-Letter-Warteschlangen erforderlich
Häufige Anti-Patterns
Anti-Pattern 1: Keine Dead-Letter-Warteschlange
Anfängerdenken: „3 Wiederholungen versuchen, dann stillschweigend fehlschlagen"
Problem: Fehlgeschlagene Jobs verschwinden ohne Sichtbarkeit oder Wiederherstellungspfad.
Richtiger Ansatz:
// BullMQ mit Dead Letter Queue
const queue = new Queue('email-queue', {
connection: redis,
defaultJobOptions: {
attempts: 3,
backoff: {
type: 'exponential',
delay: 2000
},
removeOnComplete: 100, // Letzte 100 erfolgreiche behalten
removeOnFail: false // Alle fehlgeschlagenen zur Prüfung behalten
}
});
// Fehlgeschlagene Jobs überwachen
const failedJobs = await queue.getFailed();
Zeitstrahl:
- Vor 2020: Wiederholen und vergessen
- 2020+: Dead-Letter-Warteschlangen Standard
- 2024+: Observability für Job-Fehlschläge erforderlich
Anti-Pattern 2: Synchrone Jobverarbeitung
Symptom: API-Endpunkt wartet auf Job-Abschluss
Problem:
// ❌ FALSCH - Blockiert API-Antwort
app.post('/send-email', async (req, res) => {
await sendEmail(req.body.to, req.body.subject);
res.json({ success: true });
});
Warum falsch: Timeout, schlechte Benutzererfahrung, Serverressourcenverschwendung
Richtiger Ansatz:
// ✅ RICHTIG - In die Warteschlange einreihen und sofort zurückgeben
app.post('/send-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'
});
});
// Separater Worker verarbeitet den Job
worker.process('send', async (job) => {
await sendEmail(job.data.to, job.data.subject);
});
Anti-Pattern 3: Keine Idempotenz
Problem: Job läuft zweimal → doppelte Belastungen, doppelte E-Mails
Warum es passiert:
- Redis-Verbindung bricht während der Verarbeitung ab
- Worker stürzt vor Job-Abschluss ab
- Job-Timeout löst Wiederholung aus, während er noch läuft
Richtiger Ansatz:
// ✅ Idempotenter Job mit Deduplizierungsschlüssel
await queue.add('charge-payment', {
userId: 123,
amount: 50.00
}, {
jobId: `payment-${orderId}`, // Verhindert Dubletten
attempts: 3
});
// Im Worker: Prüfen, ob bereits verarbeitet
worker.process('charge-payment', async (job) => {
const { userId, amount } = job.data;
// Idempotenz prüfen
const existing = await db.payments.findOne({
jobId: job.id
});
if (existing) {
return existing; // Bereits verarbeitet
}
// Zahlung verarbeiten
const result = await stripe.charges.create({...});
// Idempotenz-Datensatz speichern
await db.payments.create({
jobId: job.id,
result
});
return result;
});
Anti-Pattern 4: Kein Rate Limiting
Problem: Überlastung von Drittanbieter-APIs oder Ausschöpfen von Kontingenten
Symptom: „Rate limit exceeded"-Fehler von Sendgrid, Stripe usw.
Richtiger Ansatz:
// BullMQ Rate Limiting
const queue = new Queue('api-calls', {
limiter: {
max: 100, // Maximal 100 Jobs
duration: 60000 // Pro 60 Sekunden
}
});
// Oder: Prioritätsbasierte Ratenbegrenzungen
await queue.add('send-email', data, {
priority: user.isPremium ? 1 : 10,
rateLimiter: {
max: user.isPremium ? 1000 : 100,
duration: 3600000 // Pro Stunde
}
});
Anti-Pattern 5: Worker-Skalierung vergessen
Problem: Ein einzelner Worker kann mit der Warteschlangentiefe nicht Schritt halten
Symptom: Warteschlange staut sich, Jobs haben Stunden/Tage Verzögerung
Richtiger Ansatz:
// Horizontale Skalierung mit mehreren Workern
const worker = new Worker('email-queue', async (job) => {
await processEmail(job.data);
}, {
connection: redis,
concurrency: 5 // 5 Jobs gleichzeitig pro Worker verarbeiten
});
// Mehrere Worker-Prozesse ausführen (PM2, Kubernetes usw.)
// Jeder Worker verarbeitet concurrency * num_workers Jobs
Überwachung:
// Alarme für Warteschlangentiefe einrichten
setInterval(async () => {
const waiting = await queue.getWaitingCount();
if (waiting > 1000) {
alert('Warteschlangentiefe überschreitet 1000, Worker skalieren!');
}
}, 60000);
Implementierungsmuster
Muster 1: E-Mail-Kampagnen
// Warteschlangeneinrichtung
const emailQueue = new Queue('email-campaign', { connection: redis });
// Batch einreihen
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 mit Wiederholungslogik
const worker = new Worker('email-campaign', 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; // Wiederholen
}
// Ungültige E-Mail, nicht wiederholen
console.error(`Ungültige E-Mail für Benutzer ${userId}`);
}
}, {
connection: redis,
concurrency: 10
});
Muster 2: Zeitgesteuerte Berichte
// Täglicher Bericht um 9 Uhr
await queue.add('daily-report', {
type: 'sales',
recipients: ['[email protected]']
}, {
repeat: {
pattern: '0 9 * * *', // Cron-Syntax
tz: 'America/New_York'
}
});
// Worker erstellt und versendet Bericht
worker.process('daily-report', async (job) => {
const { type, recipients } = job.data;
const data = await generateReport(type);
const pdf = await createPDF(data);
await emailQueue.add('send', {
to: recipients,
subject: `Täglicher ${type}-Bericht`,
attachments: [{ filename: 'report.pdf', content: pdf }]
});
});
Muster 3: Video-Transkodierungs-Pipeline
// Mehrstufiger Job mit Fortschrittsverfolgung
await videoQueue.add('transcode', {
videoId: 123,
formats: ['720p', '1080p', '4k']
}, {
attempts: 2,
timeout: 3600000 // 1 Stunde Timeout
});
worker.process('transcode', async (job) => {
const { videoId, formats } = job.data;
for (let i = 0; i < formats.length; i++) {
const format = formats[i];
// Fortschritt aktualisieren
await job.updateProgress((i / formats.length) * 100);
// Transkodieren
await ffmpeg.transcode(videoId, format);
}
await job.updateProgress(100);
});
// Client fragt Fortschritt ab
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
});
});
Überwachung & Observability
Wesentliche Metriken
// Dashboard zur Warteschlangengesundheit
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, die auf Verarbeitung warten
active, // Jobs, die gerade verarbeitet werden
completed, // Erfolgreich abgeschlossen
failed, // Fehlgeschlagen nach Wiederholungen
delayed, // Für die Zukunft geplant
health: waiting < 1000 && failed < 100 ? 'healthy' : 'degraded'
};
}
BullMQ Board (UI)
// Entwicklung: Jobs visuell überwachen
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());
// Besuchen Sie http://localhost:3000/admin/queues
Produktions-Checkliste
□ Dead-Letter-Warteschlange konfiguriert
□ Wiederholungsstrategie mit exponentiellem Backoff
□ Job-Timeout-Grenzen gesetzt
□ Rate Limiting für Drittanbieter-APIs
□ Idempotenzschlüssel für kritische Operationen
□ Worker-Parallelität optimiert (CPU-Kerne * 2)
□ Horizontale Skalierung konfiguriert (mehrere Worker)
□ Überwachung der Warteschlangentiefe mit Alarmen
□ Workflow zur Inspektion fehlgeschlagener Jobs
□ Job-Daten enthalten keine personenbezogenen Daten in Logs
□ Redis-Persistenz aktiviert (AOF oder RDB)
□ Graceful-Shutdown-Handling (SIGTERM)
Verwendung vs. Vermeidung
| Szenario | Hintergrundjobs verwenden? |
|---|---|
| Begrüßungs-E-Mail bei Anmeldung senden | ✅ Ja - kann 2-5 Sekunden dauern |
| Kreditkarte belasten | ⚠️ Vielleicht - abhängig von der Latenz des Zahlungsanbieters |
| PDF-Bericht erstellen (30 Sekunden) | ✅ Ja - definitiv Hintergrund |
| Benutzerprofil aus DB abrufen | ❌ Nein - Millisekunden, synchron bleiben |
| Video-Upload verarbeiten (5 Minuten) | ✅ Ja - immer Hintergrund |
| Formulareingabe validieren | ❌ Nein - synchrone Validierung |
| Täglicher Cron-Job | ✅ Ja - wiederholbare Jobs verwenden |
| Echtzeit-Chat-Nachricht | ❌ Nein - WebSockets verwenden |
Technologievergleich
| Funktion | BullMQ | Celery | AWS SQS |
|---|---|---|---|
| Sprache | Node.js | Python | Beliebig (HTTP-API) |
| Backend | Redis | Redis/RabbitMQ/SQS | Verwaltet |
| Prioritäten | ✅ | ✅ | ✅ |
| Rate Limiting | ✅ | ❌ | ✅ (über Attribute) |
| Wiederholung/Cron | ✅ | ✅ (celery-beat) | ❌ (EventBridge verwenden) |
| UI-Dashboard | Bull Board | Flower | CloudWatch |
| Workflows | ❌ | ✅ (Chains, Groups) | ❌ |
| Lernkurve | Mittel | Mittel | Niedrig |
| Kosten | Redis-Hosting | Redis-Hosting | 0,40 $/Millionen Anfragen |
Referenzen
/references/bullmq-patterns.md- Fortgeschrittene BullMQ-Muster und Beispiele/references/celery-workflows.md- Celery Chains, Groups und Chords/references/job-observability.md- Überwachung, Alarmierung und Debugging
Skripte
scripts/setup_bullmq.sh- BullMQ mit Redis initialisierenscripts/queue_health_check.ts- Dashboard für Warteschlangenmetrikenscripts/retry_failed_jobs.ts- Massenweise fehlgeschlagene Jobs wiederholen
Dieses Skill leitet: Hintergrundjob-Implementierung | Warteschlangenarchitektur | Wiederholungsstrategien | Worker-Skalierung | Job-Observability