Eine einfache Job-Queue mit Redis bauen

Eine Willkommensmail verschicken, ein hochgeladenes Bild skalieren, einen PDF-Bericht erzeugen – nichts davon sollte innerhalb der HTTP-Anfrage passieren, die es ausgelöst hat. Eine Job-Queue erlaubt es, die Arbeit zu übergeben und dem Nutzer sofort zu antworten, und Redis bringt gerade genug Primitive mit, um eine ziemlich solide zu bauen, ohne einen schwergewichtigeren Message Broker einzuführen.

Warum nicht einfach inline arbeiten

Die naive Version erledigt die langsame Arbeit direkt im Request-Handler:

app.post('/signup', async (req, res) => {
  const user = await createUser(req.body);
  await sendWelcomeEmail(user); // blocks the response on an SMTP round trip
  res.json(user);
});

Ist der E-Mail-Anbieter langsam oder kurz nicht erreichbar, hängt oder scheitert jede Anmeldung, obwohl das Anlegen des Nutzers geklappt hat. Eine Queue entkoppelt „der Nutzer wurde angelegt" von „die E-Mail wurde verschickt" – die Anfrage kehrt schnell zurück, und ein separater Worker-Prozess übernimmt die Zustellung, mit Retries, unabhängig davon.

Das Kernprimitiv: ein blockierendes Listen-Pop

Redis-Listen unterstützen LPUSH (vorne hinzufügen) und BRPOP (blockieren und von hinten entnehmen, wartend, wenn die Liste leer ist). Dieses Paar ist eine Queue:

import { createClient } from 'redis';

const redis = createClient();
await redis.connect();

// Producer: enqueue a job
async function enqueue(queueName, job) {
  await redis.lPush(queueName, JSON.stringify(job));
}

// Consumer: block until a job is available, then process it
async function dequeue(queueName) {
  const result = await redis.brPop(queueName, 0); // 0 = block indefinitely
  return JSON.parse(result.element);
}

Verwendung:

await enqueue('emails', { type: 'welcome', userId: user.id });

// worker.js — runs as a separate process
while (true) {
  const job = await dequeue('emails');
  await sendWelcomeEmail(job.userId);
}

Das funktioniert, hat aber eine ernste Lücke: Stürzt der Worker ab, nachdem BRPOP den Job aus der Liste entfernt hat, aber bevor er die Verarbeitung abgeschlossen hat, ist der Job weg. BRPOP ist in dem Moment destruktiv, in dem es zurückkehrt – nichts hält fest, dass der Job je abgeholt wurde.

Zuverlässig machen: das Reliable-Queue-Muster

Die Redis-Dokumentation selbst beschreibt dafür ein Muster: den Job atomar aus der Hauptqueue in eine „Processing"-Liste pro Worker verschieben, sodass er nie in einem Zustand ist, in dem er nirgends existiert.

async function dequeueReliable(queueName, processingKey) {
  const raw = await redis.brPopLPush(queueName, processingKey, 0);
  return raw ? JSON.parse(raw) : null;
}

async function ack(processingKey, rawJob) {
  await redis.lRem(processingKey, 1, rawJob);
}

while (true) {
  const raw = await redis.brPopLPush('emails', 'emails:processing', 0);
  try {
    const job = JSON.parse(raw);
    await sendWelcomeEmail(job.userId);
    await redis.lRem('emails:processing', 1, raw);
  } catch (err) {
    console.error('job failed, leaving in processing list for recovery', err);
  }
}

Stirbt der Worker mitten im Job, bleibt der Job in emails:processing liegen, statt zu verschwinden. Ein separater Wiederherstellungsprozess – periodisch oder beim Start eines Workers ausgeführt – kann diese Liste nach Jobs durchsuchen, die älter als ein Schwellenwert sind, und sie für einen weiteren Versuch zurück in die Hauptqueue schieben.

Retry mit Backoff hinzufügen

Ein fehlgeschlagener Job sollte nicht sofort wieder an die Reihe kommen und in einer engen Schleife erneut scheitern. Verfolge einen Versuchszähler in der Job-Nutzlast und reihe verzögert neu ein:

async function handleJob(raw, processingKey) {
  const job = JSON.parse(raw);
  try {
    await sendWelcomeEmail(job.userId);
    await redis.lRem(processingKey, 1, raw);
  } catch (err) {
    await redis.lRem(processingKey, 1, raw);
    job.attempts = (job.attempts ?? 0) + 1;

    if (job.attempts >= 5) {
      await redis.lPush('emails:dead-letter', JSON.stringify(job));
      return;
    }

    const delayMs = 1000 * 2 ** job.attempts;
    setTimeout(() => enqueue('emails', job), delayMs);
  }
}

Beachte die Dead-Letter-Queue: Nach genug fehlgeschlagenen Versuchen hört man auf, automatisch zu wiederholen, und verschiebt den Job an einen Ort, an dem ein Mensch ihn untersuchen kann. Ohne das wiederholt sich ein dauerhaft kaputter Job (fehlerhafte Daten, ein nachgelagerter Dienst, der endgültig weg ist) ewig und ertränkt echte Arbeit in den Logs.

Geplante / verzögerte Jobs mit einem Sorted Set

LPUSH/BRPOP deckt nur „so bald wie möglich ausführen" ab. Für „führe das zu einem bestimmten zukünftigen Zeitpunkt aus" – eine Erinnerungsmail, eine Prüfung der Abo-Verlängerung – funktioniert ein Sorted Set besser, mit dem geplanten Zeitstempel als Score:

async function scheduleAt(queueName, job, runAtMs) {
  await redis.zAdd(`${queueName}:scheduled`, {
    score: runAtMs,
    value: JSON.stringify(job),
  });
}

Ein Poller verschiebt fällige Jobs periodisch in die reguläre Queue:

async function promoteDueJobs(queueName) {
  const now = Date.now();
  const due = await redis.zRangeByScore(`${queueName}:scheduled`, 0, now);

  for (const raw of due) {
    await redis.multi()
      .zRem(`${queueName}:scheduled`, raw)
      .lPush(queueName, raw)
      .exec();
  }
}

setInterval(() => promoteDueJobs('emails'), 1000);

Die multi()-Transaktion ist hier wichtig – sie stellt sicher, dass ein Job nie aus dem Set der geplanten Jobs entfernt wird, ohne auch in der aktiven Queue zu landen, selbst wenn der Prozess zwischen den beiden Operationen unterbrochen wird.

Wann man aufhören sollte, das selbst zu bauen

Damit hat man eine wirklich nützliche Queue, aber Produktions-Job-Systeme brauchen außerdem Dinge, die richtig Aufwand kosten, wenn sie korrekt sein sollen: Job-Prioritäten, Concurrency-Limits pro Queue, ein Dashboard zum Untersuchen hängender Jobs, Zustellgarantien nahe „exactly-once" bei Redis-Failover. Dann greift man zu einer gepflegten Bibliothek – BullMQ ist die Standardwahl im Node-Ökosystem und implementiert alle oben genannten Muster (und mehr) auf Redis. Eine minimale Version zuerst selbst zu bauen lohnt sich aber einmal: Es ist der schnellste Weg zu verstehen, wofür eine Job-Queue tatsächlich zuständig ist, sodass man beurteilen kann, ob die Standardwerte einer Bibliothek zu den eigenen Fehlerszenarien passen, statt ihnen blind zu vertrauen.