Ereignisströme
Kurz gesagt: Ein Ereignisstrom wie Kafka ist ein Protokoll, in das jedes Ereignis einmal geschrieben wird und das jedes System in seinem eigenen Tempo liest.
Problem
Eine Bestellung, vier Systeme
Kunde 7 gibt eine Bestellung über 12,40 € auf. Das Lager muss sie packen, das Rechnungsteam eine Rechnung ausstellen, der E-Mail-Dienst eine Nachricht senden, und die Analyse den Verkauf zählen.
Ruft der Code der Bestellung die vier Systeme nacheinander auf, lässt ein Mailserver, der nicht antwortet, die ganze Bestellung scheitern. Oder die Bestellung geht durch, und die E-Mail geht verloren. Jedes neue System (sagen wir Treuepunkte) bedeutet eine Änderung im Code der Bestellung.
Die Tabelle orders hält, was die Bestellung jetzt ist. Sie hält nicht, wer schon davon weiß und was vorher geschah. Du brauchst einen Ort, an dem der Shop „die Bestellung wurde aufgegeben“ einmal sagt, und jeder, den es betrifft, liest es, wenn er kann.
Die Idee
Ein Protokoll von Ereignissen
Ein Ereignis ist etwas, das schon geschehen ist: „Bestellung 61 wurde aufgegeben“, „Bestellung 61 wurde bezahlt“. Der Shop schreibt es ans Ende eines Protokolls und vergisst es. Das Lager, die Rechnungen und die E-Mail lesen das Protokoll jeweils in ihrem eigenen Tempo. Ist der E-Mail-Dienst eine Stunde lang aus, macht er dort weiter, wo er aufgehört hat, und nichts geht verloren.
Ein solches Protokoll ist ein Ereignisstrom. Kafka hält ihn in Topics, zum Beispiel orders. Ein Topic ist in Partitionen aufgeteilt, damit viele Maschinen gleichzeitig schreiben und lesen können. Jeder Eintrag bekommt in seiner Partition die nächste Nummer: den Offset. Die Nummern steigen nur.
produce schreibt ein Ereignis und sagt dir, in welche Partition es ging und unter welcher Nummer. read zeigt alles: p1:0 ist Partition 1, Offset 0. Im Bild ist jede Zeile eine Partition, die Zellen sind die Einträge mit ihren Nummern (unter der Nummer steht das Ende des Schlüssels), und die gestrichelte Zelle ist die Stelle, an die der nächste Eintrag kommt.
Einträge werden nicht verändert oder gelöscht: Befehle wie update und delete liefern einen Fehler. Wird etwas korrigiert, ist das ein neues Ereignis.
Genauer, für Neugierige
Die Befehle des Simulators sind eine besser lesbare Version der Kafka-Werkzeuge (kafka-topics, kafka-console-producer, kafka-console-consumer, kafka-consumer-groups):
create topic orders partitions 3: ein neues Topic mit 3 Partitionen;topicslistet die Topics auf;produce orders key=customer-7 {"order":61,"total":12.40}: schreibt ein Ereignis; der Schlüssel ist optional;read orders [partition 1] [from 0]: zeigt die Einträge und merkt sich nichts;consume orders group billing [as billing-2] [max 5] [from-beginning] [no-commit]: liest als Gruppe und speichert, wie weit sie gekommen ist;offsets orders: das Ende jeder Partition und die Position jeder Gruppe;describe group billing: die Position, das Ende und der Rückstand der Gruppe in jeder Partition;groupslistet die Gruppen auf;reset group billing to 0(oderearliest,latest): verschiebt die Position der Gruppe.
no-commit gibt es nur im Simulator: Es zeigt einen Consumer, der gelesen hat, aber abgestürzt ist, bevor er seine Position gespeichert hat. Der Simulator liest Partition für Partition (erst 0, dann 1, dann 2). Ein echter Consumer holt Pakete aus mehreren Partitionen gleichzeitig, und die Reihenfolge dazwischen ist nicht festgelegt.
Sieh es dir an
Zwischen Partitionen geht die Reihenfolge verloren
Ohne Schlüssel werden die Einträge auf die Partitionen verteilt (der Simulator nimmt sie reihum: 0, 1, 2, 0). Bestellung 61 wird aufgegeben, bezahlt und versandt. Der E-Mail-Dienst liest:
Der E-Mail-Dienst sieht „shipped“ für Bestellung 61 vor „placed“ und schickt „dein Paket ist unterwegs“ für eine Bestellung, die er noch nicht kennt. Die Reihenfolge ist nur innerhalb einer Partition garantiert.
Dafür ist der Schlüssel da. Das Programm, das schreibt, berechnet aus dem Schlüssel eine Zahl (einen Hash) und nimmt den Rest der Division durch die Anzahl der Partitionen. Derselbe Schlüssel ergibt immer dieselbe Partition. Der Hash des Simulators ist die Summe der Zeichencodes: Für customer-7 ist er 982, und 982 geteilt durch 3 lässt den Rest 1, also Partition 1. Kafka nutzt einen anderen Hash (murmur2), aber die Idee ist dieselbe.
Der Schlüssel ist das, dessen Reihenfolge erhalten bleiben muss: die Bestellnummer für die Ereignisse einer Bestellung, oder der Kunde, wenn die Reihenfolge aller seiner Bestellungen zählt.
Probier es aus
Die Ereignisse einer Bestellung
Das Topic orders hat 3 Partitionen. Schreibe die drei Ereignisse von Bestellung 61 der Reihe nach (placed, paid, shipped) mit dem Schlüssel order-61, damit jeder Consumer sie in dieser Reihenfolge liest.
Lösung
produce orders key=order-61 {"order":61,"event":"placed"}
produce orders key=order-61 {"order":61,"event":"paid"}
produce orders key=order-61 {"order":61,"event":"shipped"}
Dreimal produce orders key=order-61 …. Alle drei gehen in eine Partition (1) unter den Nummern 0, 1 und 2, und jeder Consumer liest sie der Reihe nach. Ohne Schlüssel gehen sie in drei verschiedene Partitionen, und die Reihenfolge geht verloren. Der Schlüssel bewahrt die Reihenfolge, in der du sie schreibst: Schreibst du „shipped“ vor „paid“, werden sie so gelesen.
Die Idee
Consumer-Gruppen
Jedes System liest als Consumer-Gruppe: billing, email, warehouse. Die Gruppe merkt sich, wie weit sie in jeder Partition gekommen ist. Nach dem Lesen speichert sie diese Position (ein Commit), sodass sie auch nach einem Neustart dort weitermacht. Zwei Gruppen kommen sich nicht in die Quere: Jede bekommt jedes Ereignis.
billing hat alles gelesen, email nur zwei Einträge. Der Unterschied zwischen dem Ende der Partition und der Position der Gruppe ist der Rückstand (lag). Im Bild hat jede Gruppe unter jeder Partition ihre eigene Zeile: Der gefärbte Teil ist gelesen, und ▲ ist die Stelle, an der sie als Nächstes liest. Ein wachsender Rückstand heißt, dass die Gruppe nicht mithalten kann.
Innerhalb einer Gruppe werden die Partitionen aufgeteilt. Ist die Rechnungsstellung langsam, startest du eine zweite Kopie mit derselben Gruppe, und jede Kopie liest ihre eigenen Partitionen:
Deshalb ist die Anzahl der Partitionen die Obergrenze: Bei 3 Partitionen steht eine vierte Kopie untätig da.
Eine neue Gruppe ohne from-beginning beginnt am Ende: Sie sieht nur die Ereignisse von jetzt an. from-beginning gilt nur für eine Gruppe, die noch keine gespeicherte Position hat.
Das Analyseteam fängt heute an und will alle bisherigen Bestellungen. Lies sie mit einer neuen Gruppe analytics.
Lösung
consume orders group analytics from-beginning
consume orders group analytics from-beginning. Die neue Gruppe hat keine Position und beginnt beim ersten Eintrag. Ohne from-beginning beginnt sie am Ende und liest nichts. Und mit der Gruppe billing würde die Analyse die Position der Rechnungen übernehmen: Diese Gruppe hat schon alles gelesen.
Sieh es dir an
Zwei Rechnungen für eine Bestellung
Die Gruppe speichert ihre Position, nachdem sie gelesen hat. Stürzt der Consumer zwischen der Verarbeitung und dem Speichern ab, bekommt er nach dem Neustart dieselben Ereignisse noch einmal:
Die Bestellungen 64 und 61 kommen zweimal. Schickt billing bei jedem Lesen eine Rechnung, bekommt der Kunde zwei. Deshalb muss der Consumer idempotent sein: Dasselbe Ereignis ein zweites Mal anzuwenden ändert nichts. Am häufigsten lässt man die Datenbank das zweite ablehnen:
UNIQUE auf der Bestellnummer und ON CONFLICT DO NOTHING machen das zweite Ereignis harmlos.
Genauer, für Neugierige
„Genau einmal“ ist schwer. Die andere Reihenfolge (erst die Position speichern, dann verarbeiten) ergibt „höchstens einmal“: Bei einem Absturz geht das Ereignis verloren. Üblich ist deshalb „mindestens einmal“ plus ein idempotenter Consumer. Kafka hat einen idempotenten Producer und Transaktionen, die „genau einmal“ für das Lesen und Schreiben innerhalb von Kafka selbst bewahren. Eine E-Mail, die rausgegangen ist, kann keine Transaktion zurückholen.
Die Idee
Noch einmal von vorn
Der Bericht der Analyse war falsch: Er hat die Liefergebühr vergessen. Die Korrektur ist fertig, aber die alten Bestellungen sind schon gelesen. Das Protokoll hält die Ereignisse, deshalb kann die Gruppe zurückgehen und sie noch einmal lesen (Replay):
reset group analytics to 0 verschiebt die Position der Gruppe an den Anfang jeder Partition, und das nächste Lesen geht wieder durch alles. Die anderen Gruppen merken nichts. from-beginning hilft hier nicht: Die Gruppe hat schon eine Position, und die Position gewinnt.
Replay ist nur für einen idempotenten Consumer sicher. Ein Bericht, der neu berechnet wird, ist in Ordnung. Rechnungen, die erneut gesendet werden, nicht.
Einträge bleiben trotzdem nicht für immer. Ein Topic hat eine Aufbewahrungsdauer (Retention): Ereignisse werden gelöscht, wenn sie zu alt werden (in Kafka standardmäßig nach 7 Tagen) oder wenn sie eine festgelegte Größe überschreiten. Das Alter löscht sie, nicht die Consumer: Was alle gelesen haben, bleibt bis dahin, und was niemand gelesen hat, ist danach weg. Der Simulator bildet die Retention nicht nach.
Genauer, für Neugierige
Im echten Kafka braucht kafka-consumer-groups --reset-offsets, dass die Gruppe angehalten ist, und --execute, sonst zeigt es nur, was es tun würde. Es gibt auch kompaktierte Topics (Log Compaction): Für jeden Schlüssel bleibt mindestens der letzte Eintrag erhalten, und ältere Einträge mit demselben Schlüssel werden gelöscht.
Die Idee
Der Zustand ist die Summe der Ereignisse
In der Tabelle products ist der Bestand an Weißkäse eine einzige Zahl:
Die Zahl ist 18, aber sie sagt nicht, warum. Wie viel kam herein, wie viel wurde verkauft, wurde etwas zurückgegeben? UPDATE überschreibt den alten Wert, und die Geschichte ist weg.
Event Sourcing hält die Ereignisse als Wahrheit, und der Zustand wird aus ihnen berechnet: Der Bestand ist die Summe aller Änderungen. Dasselbe in SQL, mit dem Bestand nach jedem Ereignis:
Und so sieht es im Strom aus. Der Schlüssel ist das Produkt, damit alle seine Ereignisse in einer Partition und in Reihenfolge liegen:
Die Ereignisse liefern dir die Geschichte umsonst: wer was wann getan hat, den Bestand an jedem vergangenen Tag und eine neue Berechnung über die alten Ereignisse. Der Preis ist, dass der aktuelle Zustand ein Durchgehen aller Ereignisse bedeutet. Deshalb halten Programme auch einen fertigen Zustand (eine Momentaufnahme), den sie aus dem Strom aktualisieren. Die Ereignisse bleiben für immer, deshalb muss sich auch ihr Format vorsichtig ändern.
Übung
Zurückspulen, eine zweite Kopie und eine Korrektur
1. Die Gruppen billing und analytics haben alles in orders gelesen.
Die Berechnung der Analyse ist korrigiert. Verschiebe die Gruppe analytics zurück an den Anfang von orders, damit sie alles noch einmal liest. Fass billing nicht an und lies noch nicht.
Lösung
reset group analytics to 0
reset group analytics to 0 (oder to earliest). offsets orders zeigt analytics in jeder Partition bei 0 und billing am Ende. consume … from-beginning hilft nicht: Die Gruppe hat schon eine Position. reset group billing to 0 würde die Rechnungen noch einmal senden.
2. billing hat alles gelesen, und dann kommt Bestellung 65 (in Partition 2).
Starte eine zweite Kopie der Rechnungsstellung namens billing-2 in derselben Gruppe, damit sie einen Teil der Partitionen übernimmt und liest, was darin neu ist.
Lösung
consume orders group billing as billing-2
consume orders group billing as billing-2. Die beiden Kopien teilen sich die Partitionen: billing-1 nimmt 0 und 1, und billing-2 nimmt 2 und liest Bestellung 65. group billing-2 ist eine andere Gruppe: Sie würde jedes Ereignis gesondert bekommen, und jeder Kunde bekäme zwei Rechnungen.
3. Das Topic stock hat die Ereignisse des Weißkäses (Produkt 7): eine Lieferung von 24, Verkäufe von 4 und 3.
Der Kunde aus Bestellung 61 hat einen Käse zurückgegeben. Füge das Ereignis {"product":7,"change":1,"why":"returned"} hinzu, damit es in der Geschichte des Käses steht, nach den Verkäufen.
Lösung
produce stock key=product-7 {"product":7,"change":1,"why":"returned"}
produce stock key=product-7 {"product":7,"change":1,"why":"returned"}. Das Ereignis geht mit den anderen zu Produkt 7 in Partition 2, unter der Nummer 3. Ohne Schlüssel geht es in eine andere Partition, und seine Reihenfolge zu den Verkäufen geht verloren. Den alten Verkauf zu ändern ist nicht möglich: Einträge werden nicht angefasst, und eine Korrektur ist ein neues Ereignis.
Die Idee
Wann ein Strom, wann nicht
Ein Ereignisstrom ist gut, wenn viele Systeme von einer Sache erfahren müssen, jedes in seinem eigenen Tempo, und wenn die Geschichte etwas wert ist: Bestellungen und Zahlungen, Lagerbewegungen, Klicks und Nutzer-Tracking, Server-Logs. Er ist nicht gut, wenn du sofort eine Antwort brauchst („ist Käse auf Lager?“, „wie hoch ist die Summe?“), zum Suchen oder um einen Eintrag zu ändern. Dafür ist die SQL-Datenbank da, und sie steht meist neben dem Strom und wird aus ihm gefüllt.
Genauer, für Neugierige
Kafka ist am weitesten verbreitet: Topics, Partitionen, Gruppen, und die Positionen der Gruppen werden in Kafka selbst gehalten (im Topic __consumer_offsets). Redpanda spricht dasselbe Protokoll wie Kafka, deshalb funktionieren Kafka-Programme und -Werkzeuge damit. RabbitMQ ist in erster Linie eine Warteschlange, in der eine Nachricht verschwindet, sobald sie bestätigt ist. Seine Streams sind eine eigene Art von Warteschlange: ein Protokoll, das ab jeder Position und noch einmal gelesen werden kann. Amazon Kinesis Data Streams nennt die Partitionen Shards und hält Einträge standardmäßig 24 Stunden.
Innerhalb einer Browserseite ist dieselbe Idee ein Event-Bus (Publish/Subscribe): Ein Button meldet „in den Warenkorb gelegt“, und der Zähler und die Summe abonnieren das. Der Unterschied ist, dass ein Bus nichts aufbewahrt: Wer in dem Moment nicht zuhört, verpasst das Ereignis. Ein Protokoll bewahrt die Ereignisse und lässt dich sie noch einmal lesen.
Test