Postgres als Message Broker: Warum wir Kafka nie gebraucht haben

Wie pg_notify, Django Channels und WebSockets in vier klaren Stufen echte Echtzeit liefern — ohne Polling, ohne extra Broker, ohne Overhead.

Das Problem mit Polling

Du kennst das Muster. Daten ändern sich in der Datenbank. Der Browser soll es wissen. Also schreibt man einen setInterval, der alle paar Sekunden /api/data/ anfrägt. Funktioniert. Skaliert schlecht. Fühlt sich falsch an.

Der klassische nächste Schritt: ein Message Broker. Celery, RabbitMQ, Kafka — irgendwas, das Events weiterleitet. Mehr Moving Parts. Mehr Infrastruktur. Mehr, das kaputt gehen kann, gewartet werden muss, lizenziert oder bezahlt werden will.

Was, wenn die Lösung schon seit Jahren in eurer Datenbank schlummert?

pg_notify: schon drin, kaum bekannt

Postgres hat seit Version 9 eine eingebaute Pub/Sub-Mechanik: LISTEN und NOTIFY. Ein Trigger feuert pg_notify, ein dedizierter Listener-Prozess empfängt es asynchron — kein Polling, kein zusätzlicher Message Broker, in unter einer Millisekunde.

Das ist der gesamte Trigger:

PERFORM pg_notify(
    TG_TABLE_NAME || '_change',
    json_build_object('action', TG_OP, 'id', NEW.id)::text
);

Keine Rohdaten im Payload — das wäre riskant. Die Grenze liegt bei 8 KB; wer Row-Daten mitschickt, baut eine tickende Zeitbombe ein. Nur Aktion und ID. Die Daten holt sich, wer sie braucht, selbst nach.

Fünf Zeilen SQL. Das ist der gesamte Einstieg in die Echtzeit-Architektur.

Die Architektur: vier Stufen, eine Aufgabe je

Datenpfad — Trigger bis Browser Postgres Trigger · pg_notify NOTIFY psycopg3 Listener · 50 ms Batch group_send Redis channel_layer entities_changed Django Channels Consumer SQL · JSON WebSocket msg WebSocket React · Redux Redux dispatch

Jede Stufe hat genau eine Aufgabe. Keine Vermischung von Fetch-Logik und Relay-Logik, kein "macht mal alles auf einmal"-Prozess. Das ist der Kern, warum das System stabil läuft.

Stufe 1 — Der Listener: dünn und schnell

Der Listener läuft als eigenständiger Django Management Command (run_pg_listener), gemanagt von supervisord — unabhängig vom ASGI-Prozess. Er hält zwei persistente psycopg3-Verbindungen: eine ausschliesslich für LISTEN / notifies(), die niemals für Queries verwendet wird. Das ist kein Luxus — psycopg3 blockiert die Connection während notifies() wartet. Eine gemeinsame Connection für Queries erzeugt Race Conditions, die schwer zu reproduzieren und noch schwerer zu debuggen sind.

Jedes NOTIFY landet in einer asyncio.Queue. Eine Coroutine wartet auf das erste Item, schläft 50 Millisekunden, leert dann die Queue — und dedupliziert:

async def _drain():
    item = await queue.get()          # blockiert bis erstes Event
    await asyncio.sleep(0.05)         # 50 ms Batch-Fenster

    seen = {(item['channel'], item['id']): item['action']}
    while not queue.empty():
        n = queue.get_nowait()
        seen[(n['channel'], n['id'])] = n['action']  # letzte Action gewinnt

    # ein group_send pro Entity-Typ
    await channel_layer.group_send('resource_plan', {
        'type':    'entities_changed',
        'channel': 'resource_assetassignment_change',
        'upserts': [id for (ch, id), act in seen.items() if act != 'DELETE'],
        'deletes': [id for (ch, id), act in seen.items() if act == 'DELETE'],
    })

Ein Bulk-Import von 500 Assignments erzeugt nicht 500 WebSocket-Nachrichten. Das 50-ms-Fenster kollabiert sie zu einem einzigen Batch. Der Browser merkt davon nichts — ausser dass alles sofort da ist. Keine SQL-Queries im Listener. Keine Businesslogik. Nur weiterleiten.

Stufe 2 — Der Django Channels Consumer: holt, was der Client braucht

Der Consumer empfängt die Batch-Nachricht vom channel_layer und macht eine einzige Raw-SQL-Query pro Entity-Typ — direkt in Postgres, mit json_build_object:

@database_sync_to_async
def _fetch_json(sql, ids):
    with connection.cursor() as cursor:
        cursor.execute(sql, [ids])         # WHERE id = ANY(%s)
        return [row[0] for row in cursor.fetchall()]

async def entities_changed(self, event):
    upsert_ids = event.get('upserts', [])
    delete_ids = event.get('deletes', [])

    rows = await _fetch_json(ASSIGNMENT_SQL, upsert_ids)

    await self.send_json({
        'type':    'entities_changed',
        'channel': event['channel'],
        'upserts': rows,
        'deletes': delete_ids,
    })

Postgres baut die JSON-Struktur selbst zusammen — kein Django ORM-Overhead, kein Python-Feldmapping, kein Serializer. Die Daten kommen fertig formatiert aus der Datenbank, direkt in die WebSocket-Nachricht.

Warum nicht im Listener fetchen? Weil N Browser-Clients nicht alle dieselben Queries machen sollen. Der Listener schickt IDs an den channel_layer. Die verbundenen Consumers fetchen einmal; Postgres antwortet einmal pro Consumer — nicht N-mal für N Clients, die dasselbe wollen.

Stufe 3 — Redis: nicht der Broker, sondern der Bus

Redis ist in diesem Setup kein klassischer Message Broker. Es ist der channel_layer von Django Channels — ein In-Memory-Bus, über den Listener und Consumer kommunizieren, ohne voneinander zu wissen. Der Listener hat keine Ahnung, wie viele Clients verbunden sind. Er schickt eine Nachricht an eine Gruppe. Redis verteilt sie an alle offenen Consumer. Fertig.

Das bedeutet auch: Redis kann kurz ausfallen, ohne dass Postgres oder der Listener abstürzen. Beim Reconnect läuft alles wieder an. Wer sich neu verbindet, bekommt den aktuellen State beim WebSocket-Handshake — kein Nachholen von verpassten Events nötig.

Stufe 4 — React und Redux: zwei Dispatches, fertig

Das WebSocket-Middleware im Frontend empfängt entities_changed und macht maximal zwei Redux-Dispatches:

case 'entities_changed': {
  const { channel, upserts, deletes } = msg

  if (channel === 'resource_assetassignment_change') {
    if (upserts.length) store.dispatch(upsert_assignments(upserts))
    if (deletes.length) store.dispatch(remove_assignments(deletes))
  }
  break
}

upsertMany und removeMany kommen direkt aus Redux Toolkit's createEntityAdapter. Keine eigene Merge-Logik. Der Store ist normalisiert, die UI reagiert automatisch.

Was dabei fast schiefgelaufen wäre

Was bleibt

Zero Polling. Kein Kafka, kein RabbitMQ, kein Celery Beat. Was läuft:

Die Infrastruktur war grösstenteils schon da. Was fehlte: ein Listener-Prozess von rund 100 Zeilen Python und ein Trigger von fünf Zeilen SQL. Das System läuft seit Monaten produktiv. Bulk-Imports von Hunderten von Objekten erzeugen keine Spike-Last. Reconnects funktionieren sauber.

Manchmal ist das Einfachste das Eleganteste — und es versteckt sich in der Datenbank, die schon läuft.

#architektur #django #django channels #echtzeit #postgresql #python #redis #websocket