Zum Inhalt
wannacode

Ereignisse und Warteschlangen

Kurz gesagt: Ein Ereignis sagt „das ist passiert“ und geht in eine Warteschlange, aus der jeder Abonnent es sich selbst holt; die Zustellung ist mindestens einmal, also muss der Consumer idempotent sein.

Problem

Die langsame E-Mail bremst die Zahlung

Wenn ein Kunde bezahlt, tut der Shop vier Dinge: Er belastet die Karte, reserviert die Ware im Lager, schickt eine Bestätigungs-E-Mail und verbucht die Zahlung in der Buchhaltung. Der Checkout ruft sie nacheinander auf:

card = Service("Karte", 2)
warehouse = Service("Lager", 3)
mailer = Service("E-Mail", 25)  # langsames SMTP: 25 Ticks
ledger = Service("Buchhaltung", 4)




def pay(order, total):
    start = Service.clock
    card.call(f"belastet {total} €")
    warehouse.call(f"Ware reserviert für {order}")
    mailer.call(f"Bestätigung für {order}")
    ledger.call(f"{total} € für {order}")
    waited = Service.clock - start
    print(f"✓ {order}: der Kunde wartet {waited} Ticks")

Jeder „Service“ hier ist ein Objekt mit einem Zähler: Jeder Aufruf addiert seine Ticks. Ticks sind die Zeit im Programm, statt Sekunden. Die E-Mail läuft über einen langsamen Mailserver und kostet 25 Ticks:

class Service:
    clock = 0  # die vergangenen Ticks


    def __init__(self, name, cost, failures=0):
        self.name = name
        self.cost = cost
        self.failures = failures  # aufgezeichnetes Szenario: wie oft hintereinander es ausfällt


    def call(self, what):
        Service.clock += self.cost
        if self.failures > 0:
            self.failures -= 1
            raise RuntimeError(f"{self.name} antwortet nicht")
        print(f"{self.name}: {what}")

Die erste Zahlung geht durch. Dann fällt die E-Mail einmal aus:

pay(1001, 120)
mailer.failures = 1  # die E-Mail fällt einmal aus
try:
    pay(1002, 80)
except RuntimeError as e:
    print(f"✗ 1002: {e}")
Ausgabe
Karte: belastet 120 €
Lager: Ware reserviert für 1001
E-Mail: Bestätigung für 1001
Buchhaltung: 120 € für 1001
✓ 1001: der Kunde wartet 34 Ticks
Karte: belastet 80 €
Lager: Ware reserviert für 1002
✗ 1002: E-Mail antwortet nicht

Der Kunde wartet 34 Ticks, 25 davon auf eine E-Mail, die genauso gut eine Minute später ankommen könnte. Und bei Bestellung 1002 ist die Karte belastet und die Ware reserviert, aber der Kunde sieht einen Fehler, weil der Mailserver nicht antwortet. Er klickt noch einmal auf „Bezahlen“.

Dazu kommt ein dritter Schmerz. Der Checkout kennt jeden, der auf ihn reagiert. Morgen will das Marketing Treuepunkte und die mobile App eine Benachrichtigung. Jede neue Zeile ist eine Änderung im Code des Checkouts.

Die Idee

Ein Ereignis: Das ist schon passiert

Der Checkout soll dem Lager nicht „reservieren“, der E-Mail nicht „senden“ und der Buchhaltung nicht „verbuchen“ sagen. Es genügt zu melden, was passiert ist: „Bestellung 1001 ist bezahlt“. Das ist ein Domain-Ereignis: eine Tatsache in der Sprache des Shops, in der Vergangenheit, mit einem Namen wie OrderPaid. Es trägt seine eigene ID und die Daten, die die anderen brauchen: die Bestellnummer, die Summe, die Menge. Was passiert ist, ändert sich nicht mehr.

Das Ereignis geht in eine Nachrichtenwarteschlange. Das Programm, das die Warteschlangen führt, ist ein Broker. Wer von Zahlungen erfahren will, abonniert sie. Der Broker führt für jeden Abonnenten eine eigene Warteschlange und legt in jede eine Kopie des Ereignisses. Das ist Publish/Subscribe: Der Checkout veröffentlicht, die Abonnenten empfangen, und keine Seite weiß von der anderen.

Die Idee kennst du schon aus Observer: Wer es wissen will, abonniert. Der Unterschied ist die Warteschlange in der Mitte. Observer ruft die Abonnenten sofort im selben Aufruf auf und wartet auf jeden. Hier lässt der Checkout das Ereignis in der Warteschlange und antwortet dem Kunden. Die Abonnenten holen es, wenn sie können, wie die Packer bei Producer und Consumer.

Checkout und Lager können jetzt getrennte Module sein, sogar getrennte Services, wie in der vorigen Lektion: Sie reden nur über Ereignisse miteinander.

Sieh es dir an

Der Broker im Speicher

Ein echter Broker ist ein eigenes Programm (Kafka, RabbitMQ). Hier ist er eine Klasse im Speicher mit einer Map von Warteschlangen, damit du den Mechanismus siehst. Das Ereignis:

@dataclass(frozen=True)
class OrderPaid:
    id: str
    order: int
    total: int
    qty: int
In Python ist das Ereignis eine @dataclass(frozen=True): Der Versuch, ein Feld zu ändern, löst einen Fehler aus.

Der Broker. Ein Abonnent gibt einen Namen und eine Funktion an, die das Ereignis verarbeitet:

class Broker:
    def __init__(self, network):
        self.queues = {}  # Abonnent → seine eigene Warteschlange
        self.handlers = {}
        self.network = network


    def subscribe(self, name, handler):
        self.queues[name] = deque()
        self.handlers[name] = handler


    def publish(self, event):
        for queue in self.queues.values():
            queue.append(event)  # eine Kopie für jeden Abonnenten


    # eine Runde: jeder Abonnent bekommt das erste Element seiner Warteschlange
    def deliver(self):
        any_left = False
        for name, queue in self.queues.items():
            if not queue:
                continue
            any_left = True
            try:
                self.handlers[name](queue[0])
            except RuntimeError as e:
                print(f"  ✗ {e}: bleibt in der Warteschlange")
                continue
            # das Ack geht verloren: das Ereignis bleibt und kommt erneut
            if self.network.ack_lost(name):
                continue
            queue.popleft()  # Ack: aus der Warteschlange nehmen
        return any_left

Das Wichtige steht in deliver. Der Abonnent bekommt das erste Ereignis seiner Warteschlange, aber das Ereignis verlässt die Warteschlange erst, wenn der Abonnent ohne Fehler fertig wird. Die Bestätigung „ich bin fertig“ heißt Ack (acknowledgement). Wirft der Abonnent eine Exception, gibt es kein Ack, und das Ereignis bleibt für die nächste Runde. Ein verlorenes Ack kommt aus der Klasse Network: ein aufgezeichnetes Szenario, wessen Ack verloren geht. Hier verliert sie nichts; auf der nächsten Karte schon.

class Network:
    def __init__(self, *lost):
        self.lost = list(lost)  # aufgezeichnetes Szenario: wessen Ack verloren geht


    def ack_lost(self, name):
        if name not in self.lost:
            return False
        self.lost.remove(name)
        print(f"  ✗ Ack von {name} geht verloren")
        return True

Der Checkout belastet jetzt nur noch die Karte und veröffentlicht:

card = Service("Karte", 2)




def pay(order, total, qty):
    start = Service.clock
    card.call(f"belastet {total} €")
    broker.publish(OrderPaid(f"evt-{order}", order, total, qty))
    waited = Service.clock - start
    print(f"✓ {order}: der Kunde wartet {waited} Ticks")

Zwei Bestellungen, und die E-Mail fällt einmal aus, wie im Problem:

mailer = Service("E-Mail", 25, 1)
ledger = Service("Buchhaltung", 4)
broker = Broker(Network())
warehouse = Warehouse()
broker.subscribe("Lager", warehouse.on)
broker.subscribe("E-Mail", lambda e: mailer.call(f"Bestätigung für {e.order}"))
broker.subscribe("Buchhaltung", lambda e: ledger.call(f"{e.total} € für {e.order}"))
pay(1001, 120, 2)
pay(1002, 80, 1)
while broker.deliver():
    pass
Ausgabe
Karte: belastet 120 €
✓ 1001: der Kunde wartet 2 Ticks
Karte: belastet 80 €
✓ 1002: der Kunde wartet 2 Ticks
Lager: 2 Stück für 1001, 8 übrig
  ✗ E-Mail antwortet nicht: bleibt in der Warteschlange
Buchhaltung: 120 € für 1001
Lager: 1 Stück für 1002, 7 übrig
E-Mail: Bestätigung für 1001
Buchhaltung: 80 € für 1002
E-Mail: Bestätigung für 1002

Der Kunde wartet bei beiden Bestellungen 2 Ticks. Die E-Mail ist immer noch langsam und fällt immer noch aus, aber das ist jetzt ihr eigenes Problem: Das Ereignis bleibt in ihrer Warteschlange und kommt in der nächsten Runde wieder. Lager und Buchhaltung merken nichts davon.

Schau auch auf die Reihenfolge. Die Bestätigung für 1001 geht raus, nachdem das Lager 1002 verarbeitet hat. Jede Warteschlange läuft in ihrem eigenen Tempo. Ein Abonnent darf sich also nicht darauf verlassen, dass ein anderer seinen Teil schon erledigt hat.

Die Idee

Mindestens einmal heißt manchmal zweimal

Der Broker nimmt das Ereignis erst nach dem Ack aus der Warteschlange. Was passiert, wenn das Lager die Ware ausbucht, aber das Ack im Netz verloren geht? Der Broker kann „das Lager ist ausgefallen, bevor es angefangen hat“ nicht von „das Lager war fertig, konnte es aber nicht sagen“ unterscheiden. Um keine Bestellung zu verlieren, liefert er das Ereignis erneut aus.

Das Skript verliert das Ack vom Lager einmal. Das Lager ist naiv: Jedes Ereignis bucht Ware aus:

class Warehouse:
    def __init__(self):
        self.stock = 10


    def on(self, e):
        self.stock -= e.qty
        print(f"Lager: {e.qty} Stück für {e.order}, {self.stock} übrig")
mailer = Service("E-Mail", 25, 0)
ledger = Service("Buchhaltung", 4)
broker = Broker(Network("Lager"))
warehouse = Warehouse()
broker.subscribe("Lager", warehouse.on)
broker.subscribe("E-Mail", lambda e: mailer.call(f"Bestätigung für {e.order}"))
broker.subscribe("Buchhaltung", lambda e: ledger.call(f"{e.total} € für {e.order}"))
pay(1001, 120, 2)
while broker.deliver():
    pass
Ausgabe
Karte: belastet 120 €
✓ 1001: der Kunde wartet 2 Ticks
Lager: 2 Stück für 1001, 8 übrig
  ✗ Ack von Lager geht verloren
E-Mail: Bestätigung für 1001
Buchhaltung: 120 € für 1001
Lager: 2 Stück für 1001, 6 übrig

Eine Bestellung über 2 Stück, und im Lager fehlen 4. Der Bestand stimmt nicht, und der Shop lehnt die nächste Bestellung ab, obwohl die Ware im Regal liegt.

Ein Broker kann eines von zwei Dingen versprechen:

  • höchstens einmal: Er entfernt das Ereignis, bevor der Abonnent arbeitet. Keine Duplikate, aber fällt der Abonnent aus, ist das Ereignis verloren;
  • mindestens einmal: Er entfernt es nach dem Ack. Nichts geht verloren, aber manchmal kommt es zweimal an.

„Genau einmal“ über ein Netz kann ein Broker allein nicht versprechen. Deshalb wählt fast jedes System „mindestens einmal“, und der Empfänger fängt das Duplikat ab.

Sieh es dir an

Ein idempotenter Consumer

Ein Consumer, für den dasselbe Ereignis ein zweites Mal nichts ändert, ist idempotent (idempotent consumer). Das Lager merkt sich die IDs der Ereignisse, die es schon verarbeitet hat, und verwirft das Duplikat:

class Warehouse:
    def __init__(self):
        self.stock = 10
        # die IDs der bereits verarbeiteten Ereignisse
        self.seen = set()


    def on(self, e):
        if e.id in self.seen:
            print(f"Lager: {e.id} ist ein Duplikat")
            return
        self.stock -= e.qty
        self.seen.add(e.id)
        print(f"Lager: {e.qty} Stück für {e.order}, {self.stock} übrig")

def publish(self, event):
    for queue in self.queues.values():
        queue.append(event)  # eine Kopie für jeden Abonnenten


# eine Runde: jeder Abonnent bekommt das erste Element seiner Warteschlange
def deliver(self):
    any_left = False
    for name, queue in self.queues.items():
        if not queue:
            continue
        any_left = True
        try:
            self.handlers[name](queue[0])
        except RuntimeError as e:
            print(f"  ✗ {e}: bleibt in der Warteschlange")
            continue
        # das Ack geht verloren: das Ereignis bleibt und kommt erneut
        if self.network.ack_lost(name):
            continue
        queue.popleft()  # Ack: aus der Warteschlange nehmen
    return any_left
Schritt für Schritt, in Worten
  1. Der Checkout belastet die Karte und veröffentlicht OrderPaid evt-1001: „Bestellung 1001 ist bezahlt“. Der Kunde wartet nur 2 Ticks.
  2. Der Broker legt eine Kopie in jede Warteschlange: Lager, E-Mail und Buchhaltung. Der Checkout hat schon geantwortet.
  3. In jeder Runde bekommt jeder Abonnent das erste Ereignis aus seiner eigenen Warteschlange. Es bleibt in der Warteschlange bis zum Ack.
  4. Das Lager bucht 2 Stück aus (8 übrig) und schreibt evt-1001 in seen. E-Mail und Buchhaltung erledigen ihren Teil.
  5. Die Acks von E-Mail und Buchhaltung kommen an, und das Ereignis verlässt ihre Warteschlangen. Das Ack vom Lager geht unterwegs verloren.
  6. Der Broker weiß nicht, dass das Lager die Arbeit erledigt hat, und liefert evt-1001 erneut aus. Das ist „mindestens einmal“.
  7. evt-1001 steht schon in seen, also verwirft das Lager es. Der Bestand bleibt 8, nicht 6.
  8. Diesmal kommt das Ack an, und die Warteschlange ist leer. Mindestens einmal geliefert, genau einmal verarbeitet.
mailer = Service("E-Mail", 25, 0)
ledger = Service("Buchhaltung", 4)
broker = Broker(Network("Lager"))
warehouse = Warehouse()
broker.subscribe("Lager", warehouse.on)
broker.subscribe("E-Mail", lambda e: mailer.call(f"Bestätigung für {e.order}"))
broker.subscribe("Buchhaltung", lambda e: ledger.call(f"{e.total} € für {e.order}"))
pay(1001, 120, 2)
while broker.deliver():
    pass
Ausgabe
Karte: belastet 120 €
✓ 1001: der Kunde wartet 2 Ticks
Lager: 2 Stück für 1001, 8 übrig
  ✗ Ack von Lager geht verloren
E-Mail: Bestätigung für 1001
Buchhaltung: 120 € für 1001
Lager: evt-1001 ist ein Duplikat

Der Broker liefert immer noch zweimal aus, aber das Lager bucht die Ware einmal aus. Die ID wird nach der Arbeit geschrieben: Fällt die Arbeit auf halbem Weg aus, kommt das Ereignis erneut und wird verarbeitet.

Manche Operationen sind von selbst idempotent: „der Status von 1001 ist bezahlt“ ist beim zweiten Mal dasselbe. „Bestand minus 2“ nicht. Dafür braucht man eine ID. In einem echten System ist seen eine Tabelle in derselben Datenbank wie der Bestand, und die Subtraktion und die ID werden in einer Transaktion geschrieben. Sonst ergibt ein Absturz dazwischen wieder ein falsches Ergebnis. Was passiert, wenn der Schreibvorgang in die Datenbank und das Senden des Ereignisses in verschiedenen Systemen liegen, steht in der nächsten Lektion.

Probier es aus

Sag die Ausgabe voraus

Das Lager ist idempotent. Diesmal verliert das Skript das Ack der E-Mail. Was gibt das Programm aus?

mailer = Service("E-Mail", 25, 0)
ledger = Service("Buchhaltung", 4)
broker = Broker(Network("E-Mail"))
warehouse = Warehouse()
broker.subscribe("Lager", warehouse.on)
broker.subscribe("E-Mail", lambda e: mailer.call(f"Bestätigung für {e.order}"))
broker.subscribe("Buchhaltung", lambda e: ledger.call(f"{e.total} € für {e.order}"))
pay(1001, 120, 2)
while broker.deliver():
    pass
Antwort

Probier es aus

Finde den Fehler

Ein Kollege schreibt das Lager um, und der Test mit dem verlorenen Ack zeigt trotzdem, dass die Ware zweimal ausgebucht wird. In welcher Zeile steckt der Fehler?

Wähle die Zeile mit dem Fehler.

Die Idee

Wann Ereignisse schaden

Ereignisse sind nicht umsonst. Nutze sie nicht, wenn:

  • du die Antwort sofort brauchst. „Ist es auf Lager?“ vor der Zahlung ist eine Frage, kein Ereignis. Wenn du ohne die Antwort nicht weiterkommst, ruf das andere Modul auf;
  • das System klein ist und in einem Prozess läuft. Ein Aufruf oder Observer genügt. Ein Broker ist ein weiteres Programm, das laufen, überwacht und aktualisiert werden muss.

Und kenne den Preis, wenn du sie doch nutzt:

  • der Ablauf steht nicht mehr an einer Stelle. Was nach der Zahlung passiert, kannst du nicht in einer Funktion lesen; du sammelst es bei den Abonnenten ein. Deshalb trägt jedes Ereignis eine ID, nach der du später in den Logs suchen kannst;
  • die anderen erfahren es später. Der Kunde bezahlt und öffnet sofort „Meine Bestellungen“, und die Rechnung ist noch nicht da. Das heißt Eventual Consistency. Die Anzeige sollte das ehrlich sagen: „Die Rechnung kommt innerhalb einer Minute“;
  • eine Reihenfolge wird nicht versprochen. Du hast es im Programm gesehen: Die Warteschlangen laufen im eigenen Tempo. Wenn die Reihenfolge zählt (bezahlt, dann storniert), müssen die Ereignisse einer Bestellung durch eine Warteschlange und einen Consumer gehen;
  • Duplikate sind deine Aufgabe. Jeder Abonnent muss idempotent sein.

Wie ein echter Broker im Betrieb gefahren und überwacht wird (Cluster, Platten, Monitoring), ist ein Thema des DevOps-Kurses ZeroToOps.

Genauer, für Neugierige

Echte Broker. Kafka führt die Ereignisse als Log, das jeder Abonnent ab seiner eigenen Position liest. Dort ist das Ack die gemerkte Position (der Offset). RabbitMQ hat Warteschlangen und ein Ack für jede Nachricht, wie in der Lektion. Es gibt auch Redis Streams, Amazon SQS und SNS, Azure Service Bus und Google Pub/Sub. In Spring ist ein Abonnent eine Methode mit @KafkaListener oder @RabbitListener. ASP.NET-Core-Projekte nutzen oft MassTransit. Python hat Clients wie confluent-kafka und aio-pika, und Celery ist eine Task-Warteschlange. Node.js hat kafkajs und amqplib, und NestJS hat ein Messaging-Modul.

Die Dead Letter Queue. Ein Ereignis, das jedes Mal fehlschlägt (kaputte Daten), darf nicht ewig in der Warteschlange kreisen. Nach einigen Versuchen verschiebt der Broker es in eine eigene Warteschlange, in der sich ein Mensch darum kümmert.

Genau einmal. Kafka hat einen „Exactly-once“-Modus, aber er gilt innerhalb von Kafka: lesen, verarbeiten, zurück nach Kafka schreiben. Sobald die Wirkung außerhalb liegt (eine E-Mail, eine Zahlung, eine Datenbank), bleibt das Duplikat deine Aufgabe.

Ereignis oder Befehl. OrderPaid ist ein Ereignis: viele Abonnenten, keiner verpflichtet. ReserveStock ist ein Befehl: ein Empfänger, der ihn ausführen muss. Beide können durch eine Warteschlange gehen, aber die Namen drücken verschiedene Absichten aus.

Wie es in den Sprachen aussieht. In Python ist ein Abonnent alles, was man aufrufen kann: eine Methode eines Objekts (warehouse.on) oder ein lambda. Es gibt kein Interface: Der Broker ruft auf, was du ihm gibst. deque entfernt vorn schnell, während list.pop(0) alle Elemente verschiebt.

Übung

Eine Benachrichtigung in der App

Löse die Aufgabe in deinem Editor.

Die mobile App will eine Benachrichtigung „Bestellung bezahlt“. Füge einen vierten Abonnenten „App“ hinzu, ohne den Code des Checkouts anzufassen. Er muss idempotent sein, wie das Lager. Das Skript verliert sein Ack einmal:

Ausgabe
Karte: belastet 120 €
✓ 1001: der Kunde wartet 2 Ticks
Lager: 2 Stück für 1001, 8 übrig
E-Mail: Bestätigung für 1001
Buchhaltung: 120 € für 1001
App: Benachrichtigung für 1001
  ✗ Ack von App geht verloren
App: evt-1001 ist ein Duplikat

Weitere Aufgaben

  • Eine stornierte Bestellung. Füge ein Ereignis OrderCancelled hinzu. Das Lager bucht die Ware zurück, aber nur, wenn es OrderPaid für dieselbe Bestellung verarbeitet hat. Was passiert, wenn die Stornierung vor der Zahlung ankommt?
  • Eine Dead Letter Queue. Schlägt ein Abonnent dreimal hintereinander beim selben Ereignis fehl, verschiebt der Broker es in eine Liste dead und macht mit dem nächsten weiter.
  • Treuepunkte. Füge einen Abonnenten hinzu, der für je 10 € der Summe 1 Punkt vergibt. Was muss er sich merken, um die Punkte nicht doppelt zu vergeben?

Test

Teste dich

1. Welcher Name ist ein Domain-Ereignis und kein Befehl?
2. Das Ack vom Lager geht unterwegs verloren. Was macht der Broker?
3. Warum schreibt das Lager die ID NACH dem Ausbuchen in seen und nicht davor?
4. Wann ist es besser, das andere Modul direkt aufzurufen, statt ein Ereignis zu veröffentlichen?

Erstellt von Ivo Gergov im Rahmen der Kampagne von Evo Tech für zugängliche Programmierausbildung ·