Skalierung des Outbox Patterns mit PostgreSQL - Teil 3: Echtzeit-Reaktivität und Wartung

Willkommen zum letzten Teil unserer Serie über die Skalierung des Transactional Outbox Patterns mit PostgreSQL. In Teil 2 haben wir Nebenläufigkeitsprobleme mit SKIP LOCKED gelöst. Unsere Worker können nun Ereignisse parallel verarbeiten, ohne sich gegenseitig in die Quere zu kommen.

Aber wir haben noch zwei Probleme übrig:

1. Die Polling-Steuer: Die Datenbank jede Sekunde abzufragen, ist Verschwendung, wenn keine Ereignisse vorliegen.
2. Table Bloat: Unsere Tabelle outbox_events wird im Laufe der Zeit unendlich wachsen und die Abfrageleistung verschlechtern.

Lass uns diese Probleme beheben.

Die Kosten des Pollings

Selbst mit SKIP LOCKED führt ein kontinuierliches Abfrageintervall von 1 Sekunde über 5 Worker-Instanzen hinweg zu 432.000 Abfragen pro Tag. Wenn dein System nachts größtenteils im Leerlauf ist, geben 99 % dieser Abfragen nichts zurück. Das ist eine Verschwendung von CPU, Speicher und Datenbankverbindungen.

Wir brauchen einen Weg, um zu sagen: “Hey Datenbank, sag mir Bescheid, wenn du neue Arbeit hast.”

Push-Benachrichtigungen mit LISTEN/NOTIFY

PostgreSQL verfügt über einen integrierten Publish/Subscribe-Mechanismus namens LISTEN und NOTIFY. Wir können dies nutzen, um unsere Worker sofort zu triggern, wenn eine neue Zeile eingefügt wird.

Zuerst erstellen wir einen Datenbank-Trigger, der einen NOTIFY-Befehl abfeuert, wann immer eine neue Zeile in unsere Outbox-Tabelle eingefügt wird:

CREATE OR REPLACE FUNCTION notify_outbox_event()
RETURNS trigger AS $$
BEGIN
  -- Den Kanal 'outbox_channel' mit der ID des neuen Ereignisses benachrichtigen
  PERFORM pg_notify('outbox_channel', NEW.id::text);
  RETURN NEW;
END;
$$ LANGUAGE plpgsql;

CREATE TRIGGER outbox_event_trigger
AFTER INSERT ON outbox_events
FOR EACH ROW
EXECUTE FUNCTION notify_outbox_event();

Jetzt können wir in unserem Node.js-Worker auf diese Benachrichtigung lauschen:

import { Client } from 'pg'

async function setupListener() {
  const client = new Client({ connectionString: process.env.DATABASE_URL })
  await client.connect()

  // Auf den Kanal lauschen
  await client.query('LISTEN outbox_channel')

  // Auf Benachrichtigungen warten
  client.on('notification', async (msg) => {
    console.log(`Neue Ereignisbenachrichtigung erhalten: ${msg.payload}`)
    
    // Die Funktion processOutboxQueue aus Teil 2 triggern!
    await processOutboxQueue()
  })
}

Wenn sich jetzt ein Benutzer registriert, teilt die Datenbank dem Worker sofort mit, dass er die Warteschlange verarbeiten soll. Wir haben Latenz durch Echtzeit-Reaktivität ersetzt!

Der hybride Ansatz

Du könntest versucht sein, das 1-Sekunden-Abfrageintervall komplett zu deaktivieren. Tu das nicht.

LISTEN/NOTIFY ist Fire-and-Forget. Wenn dein Node.js-Worker neu startet oder es genau in dem Moment einen temporären Netzwerkaussetzer gibt, in dem die Datenbank das NOTIFY abfeuert, ist diese Benachrichtigung für immer verloren.

Die beste Architektur ist ein hybrider Ansatz:

1. Verwende LISTEN/NOTIFY für sofortige Verarbeitung mit geringer Latenz.
2. Behalte einen langsamen Hintergrund-Poller bei (der z. B. alle 30-60 Sekunden läuft), als Sicherheitsnetz, um Ereignisse aufzufangen, die verpasst wurden oder ein Timeout hatten.

Tabellenwartung (Table Bloat)

Während dein System läuft, füllt sich die Tabelle outbox_events mit verarbeiteten Ereignissen (status = 'PUBLISHED'). Mit der Zeit wird diese riesige Tabelle deine SELECT ... FOR UPDATE SKIP LOCKED-Abfragen verlangsamen, da die Datenbank Millionen von abgeschlossenen Zeilen scannen muss, um die ausstehenden zu finden.

Du hast drei Hauptstrategien für die Bereinigung:

1. Soft Deletes vs. Hard Deletes: Du könntest einen Hintergrund-Cron-Job ausführen, der DELETE FROM outbox_events WHERE status = 'PUBLISHED' macht. Häufige DELETE-Operationen in PostgreSQL verursachen jedoch “Table Bloat” aufgrund von MVCC (Multi-Version Concurrency Control) und erfordern aggressives VACUUMing.

2. Der “Löschen bei Verarbeitung”-Ansatz: Anstatt ein Ereignis als 'PUBLISHED' zu markieren, lösche die Zeile einfach komplett innerhalb der Verarbeitungstransaktion:

   // Ereignis verarbeiten...
   // Dann löschen:
   await client.query("DELETE FROM outbox_events WHERE id = $1", [event.id])
   

   Dies ist oft der effizienteste Ansatz, wenn du keinen Audit-Trail der Ereignisse aufbewahren musst.

3. Tabellen-Partitionierung: Wenn du doch einen Audit-Trail benötigst, verwende PostgreSQL Table Partitioning (z. B. Partitionierung nach Tag oder Woche). Wenn eine Partition zu alt wird (z. B. älter als 30 Tage), kannst du einfach DROP TABLE outbox_events_2025_01 ausführen, was sofort geschieht und Speicherplatz sauber ohne Fragmentierung freigibt.

Fazit

In dieser dreiteiligen Serie haben wir eine robuste ereignisgesteuerte Architektur aufgebaut, die ausschließlich PostgreSQL verwendet:

• Wir haben Konsistenz mithilfe des Transactional Outbox Patterns innerhalb einer einzigen Transaktion garantiert.
• Wir haben sicher auf mehrere Worker skaliert, indem wir SKIP LOCKED verwendet haben.
• Wir haben Echtzeit-Verarbeitung mit geringer Latenz durch LISTEN/NOTIFY erreicht.
• Wir haben für langfristige Gesundheit mit intelligenten Bereinigungsstrategien gesorgt.

Du brauchst nicht immer eine komplexe, schwere verteilte Streaming-Plattform, um ein großartiges ereignisgesteuertes System zu bauen. Manchmal ist eine gut abgestimmte PostgreSQL-Datenbank genau das, was du brauchst.

Bleib dran und genieße jeden Schritt deiner Coding-Reise.