A einfache Job-Queue mit Redis bauen

A Willkommens-E-Mail schicken, a hochgladenes Bild skaliern, an PDF-Bericht generiern - nix davon soit in dem HTTP-Request passiern, der's ausglöst hot. Mit ana Job-Queue gibst die Orbeit ab und antwortest dem Nutzer sofort, und Redis hot grod gnua Primitive eingbaut, um a halbwegs solide zu bauen, ohne an schwereren Message Broker einzuziehn.

Warum ned einfach die Orbeit inline machn

Die naive Version mocht die langsame Orbeit 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);
});

Wenn da E-Mail-Provider langsam oda kurz down is, hängt oda scheitert jede Anmeldung, a wenn s'Erstöin vom User geklappt hot. Eine Queue entkoppelt "da User wurde angelegt" von "die E-Mail wurde gschickt" - da Request kehrt schnell zruck, und a eigener Worker-Prozess erledigt die Zustellung, mit Retries, unabhängig.

Die Kernprimitive: a blockierendes Pop auf a Liste

Redis-Listen unterstützen LPUSH (vorn hinzufügen) und BRPOP (hinten blockierend entnehmen, wartend, wenn die Liste leer is). Des Paar is a 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);
}

Des funktioniert, hot aber a ernste Lücke: Wenn da Worker abstürzt, nachdem BRPOP den Job aus da Liste entfernt hot, aber bevor er mit'm Verarbeiten fertig is, is der Job weg. BRPOP is destruktiv, sobois es zruckkehrt - es gibt ka Aufzeichnung, dass der Job je abgholt wurde.

Zuverlässig machn: Das Reliable-Queue-Muster

Redis' eigene Dokumentation beschreibt dafür a Muster: den Job atomar von da Hauptqueue in a pro-Worker-"Processing"-Liste verschieben, sodass er nie in an Zustand is, wo 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);
  }
}

Wenn da Worker mittn im Job stirbt, bleibt der Job in emails:processing liegn, statt zu verschwinden. A separater Recovery-Prozess - periodisch oda beim Worker-Start - kann die Liste nach Jobs durchsuchen, de ältern ois a Schwellenwert sand, und sie für an weiteren Versuch zruck auf de Hauptqueue schiebn.

Retry mit Backoff dazubauen

A fehlgschlogener Job soit ned einfach sofort wieder auftauchn und in ana engen Schleifn nochamoi scheitern. Track a Versuchszoi in da Job-Payload und nutz a verzögertes Wiedereinreihn:

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 fehlgschlogene Versuche hör auf, automatisch zum wiederholn, und verschieb den Job an an Ort, wo a Mensch ihn anschaun kann. Ohne des wird a dauerhaft kaputter Job (schlechte Daten, a nachgelagerter Service, der wirklich für immer weg is) ewig wiederholt und ertränkt die echte Orbeit in den Logs.

Geplante / verzögerte Jobs mit am Sorted Set

LPUSH/BRPOP behandelt nur "so bald wie möglich ausführn". Für "führ des zu am bestimmten zukünftigen Zeitpunkt aus" - a Erinnerungs-Mail, a Abo-Verlängerungs-Check - funktioniert a Sorted Set besser, mit dem geplanten Zeitstempel ois Score:

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

A Poller verschiebt periodisch fällige Jobs 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 is do wichtig - sie stellt sicher, dass a Job nie aus dem geplanten Set entfernt wird, ohne a in der aktiven Queue zu landen, a wenn der Prozess zwischen den beiden Operationen unterbrochen wird.

Wann ma aufhört, des söba zum bauen

Des gibt dir a wirklich nützliche Queue, aber Produktiv-Job-Systeme brauchn a Sochn, die echt Aufwand verlangen, um richtig zu sei: Job-Prioritäten, Concurrency-Limits pro Queue, a Dashboard zum Inspizieren hängender Jobs, Fast-genau-einmal-Zustellgarantien bei Redis-Failover. Do greif zu ana gewarteten Bibliothek - BullMQ is die Standardwahl im Node-Ökosystem und implementiert olle oben genannten Muster (und mehr) auf Redis. Zuerst a minimale Version söba zum bauen zahlt si einmal aus, weil's der schnellste Weg is zu vastehn, wofür a Job-Queue wirklich zuständig is, damit du beurteiln kannst, ob die Defaults ana Bibliothek zu deinen Fehlermodi passen, statt ihnen blind zum vertrauen.