Reaktive Systeme in Java

Kapitel 4. Entwurfsprinzipien für reaktive Systeme

Diese Arbeit wurde mithilfe von KI übersetzt. Wir freuen uns über dein Feedback und deine Kommentare: [e-mail]

In Kapitel 3 haben wir uns mit den Herausforderungen verteilter Systeme befasst. Jetzt ist es an der Zeit zu sehen, was Reactive zu bieten hat. Reactive kann als eine Reihe von Prinzipien für den Aufbau verteilter Systeme gesehen werden, eine Art Checkliste, die sicherstellt, dass bei der Architektur und dem Aufbau eines Systems kein wichtiges Problem übersehen wurde. Diese Prinzipien konzentrieren sich auf Folgendes:

Reaktionsfähigkeit

Die Fähigkeit, Anfragen zu bearbeiten, wenn Ausfälle oder Lastspitzen auftreten

Effizienz

Die Fähigkeit, mit weniger Ressourcen mehr zu erreichen

In diesem Kapitel behandeln wir die Prinzipien, die reaktive Systeme fördern.

Reaktive Systeme 101

Im Jahr 2013 versammelte sich eine Gruppe von Experten für verteilte Systeme und schrieb die erste Version von "The Reactive Manifesto". In diesem Whitepaper fassten sie ihre Erfahrungen beim Aufbau von verteilten Systemen und Cloud-Anwendungen zusammen. 2013 war die Cloud zwar noch nicht genau das, was sie heute ist, aber die dynamische Erstellung von ephemeren Ressourcen war bereits ein bekannter Mechanismus.

Das "Reactive Manifesto" definiert reaktive Systeme als verteilte Systeme mit vier Merkmalen:

Responsive

In der Lage sein, Anfragen zeitnah zu bearbeiten

Unverwüstlich

In der Lage sein, Ausfälle elegant zu bewältigen

Elastisch

Kann je nach Last und Ressourcen auf- und abwärts skaliert werden

Nachricht gesteuert

Nutzung der asynchronen, nachrichtenbasierten Kommunikation zwischen den Komponenten, die das System bilden

Diese vier Merkmale sind in Abbildung 4-1 dargestellt.

Abbildung 4-1. Merkmale reaktiver Systeme

Wenn du dieses Bild zum ersten Mal siehst, werden dich die vielen Pfeile vielleicht verwirren. Es kann wie eine maßgeschneiderte Marketingkampagne aussehen. Das ist es aber nicht, und wir wollen dir erklären, warum diese Säulen bei der Entwicklung von Cloud- und Kubernetes-nativen Anwendungen sehr sinnvoll sind. Beginnen wir mit dem unteren Teil der Abbildung.

Anstatt zu versuchen, verteilte Systeme einfacher zu machen, als sie sind, machen sich reaktive Systeme ihre asynchrone Natur zu eigen.Sie nutzen asynchrone Nachrichtenübermittlung, um das Bindegewebe zwischen den Komponenten herzustellen. Asynchrone Nachrichtenübermittlung sorgt für lose Kopplung, Isolation und Standorttransparenz. In einem reaktiven System basieren die Interaktionen auf Nachrichten, die an abstrakte Ziele gesendet werden. Diese Nachrichten übermitteln alles - sowohl Daten als auch Fehler. Asynchrone Nachrichtenübermittlung verbessert auch die Ressourcenauslastung. Durch den Einsatz von nicht blockierender Kommunikation (diesen Teil behandeln wir später in diesem Kapitel) können inaktive Komponenten fast keine CPU und keinen Speicher verbrauchen. Asynchrone Nachrichtenübermittlung ermöglicht Elastizität und Ausfallsicherheit, wie die beiden unteren Pfeile in Abbildung 4-1 zeigen.

Elastizität bedeutet, dass das System sich selbst oder Teile davon anpassen kann, um die schwankende Last zu bewältigen. Anhand der Nachrichten, die zwischen den Komponenten fließen, kann ein System feststellen, welche Teile an ihre Grenzen stoßen, und mehr Instanzen erstellen oder die Nachrichten anderswohin leiten. Die Cloud-Infrastruktur ermöglicht es, diese Instanzen zur Laufzeit schnell zu erstellen. Aber bei der Elastizität geht es nicht nur um das Hochskalieren, sondern auch um das Herunterskalieren. Das System kann entscheiden, wenig genutzte Teile zu verkleinern, um Ressourcen zu sparen. Zur Laufzeit passt sich das System selbst an und erfüllt immer die aktuelle Nachfrage, um Engpässe, Überläufe und übermäßig beanspruchte Ressourcen zu vermeiden. Wie du dir vorstellen kannst, erfordert Elastizität Beobachtbarkeit, Replikation und Routing-Funktionen. Die Beobachtbarkeit wird in Kapitel 13 behandelt. Die letzten beiden werden im Allgemeinen von der Infrastruktur wie Kubernetes oder Cloud-Providern bereitgestellt.

Resilienz bedeutet, mit Fehlern elegant umzugehen. Wie in Kapitel 3 erläutert, sind Fehler in verteilten Systemen unvermeidlich. Anstatt sie zu verstecken, betrachten reaktive Systeme Fehler als Bürger erster Klasse. Das System sollte in der Lage sein, mit ihnen umzugehen und auf sie zu reagieren. Fehler werden innerhalb jeder Komponente eingegrenzt, indem die Komponenten voneinander isoliert werden. Diese Isolierung stellt sicher, dass Teile des Systems fehlschlagen und sich erholen können, ohne das gesamte System zu gefährden. Durch die Replikation von Komponenten (Elastizität) kann das System zum Beispiel auch dann noch eingehende Nachrichten verarbeiten, wenn einige Elemente ausfallen. Die Implementierung der Ausfallsicherheit wird zwischen der Anwendung (die Ausfälle erkennen, eindämmen und, wenn möglich, elegant verarbeiten muss) und der Infrastruktur (die die Systeme überwacht und ausgefallene Komponenten neu startet) aufgeteilt.

Die letzte Eigenschaft ist der eigentliche Zweck reaktiver Systeme: Sie müssen reaktionsfähig sein. Dein System muss auch bei schwankender Last (Elastizität) und bei Ausfällen (Resilienz) reaktionsfähig bleiben, d.h. rechtzeitig reagieren. Die Weitergabe von Nachrichten ermöglicht diese Eigenschaften und noch viel mehr, wie z.B. die Flusskontrolle, indem die Nachrichten im System überwacht und bei Bedarf Gegendruck ausgeübt wird.

Kurz gesagt sind reaktive Systeme genau das, was wir bauen wollen: verteilte Systeme, die in der Lage sind, mit Unsicherheiten, Ausfällen und Last effizient umzugehen. Ihre Eigenschaften erfüllen die Anforderungen für Cloud Native und Kubernetes-native Anwendungen perfekt. Aber täusche dich nicht: Ein reaktives System zu bauen, ist immer noch ein verteiltes System. Es ist eine Herausforderung. Wenn du jedoch diese Prinzipien befolgst, wird das resultierende System reaktionsschneller, robuster und effizienter sein. Der Rest dieses Buches beschreibt, wie wir solche Systeme mit Quarkus und Messaging-Technologien einfach umsetzen können.

Befehle und Ereignisse

Nachdem wir nun viele der grundlegenden Prinzipien behandelt haben, bist du vielleicht verwirrt. In Kapitel 1 haben wir gesagt, dass reaktiv sein mit ereignisgesteuert sein zusammenhängt, aber im vorigen Abschnitt haben wir ausdrücklich asynchrone Nachrichtenübermittlung erwähnt. Bedeutet das dasselbe? Nicht ganz.

Doch zunächst müssen wir die Unterschiede zwischen Befehlen und Ereignissen erörtern. So kompliziert das Design eines verteilten Systems auch sein mag, die Konzepte von Befehlen und Ereignissen sind grundlegend. Fast alle Interaktionen zwischen einzelnen Komponenten beinhalten das eine oder das andere.

Befehle

Jedes System gibt Befehle aus.Befehle sind Aktionen, die ein Benutzer ausführen möchte. Die meisten HTTP-basierten APIs geben Befehle weiter: Der Client bittet um eine Aktion. Es ist wichtig zu verstehen, dass die Aktion noch nicht stattgefunden hat. Sie kann in der Zukunft stattfinden oder nicht; sie kann erfolgreich abgeschlossen werden oder fehlschlagen. Im Allgemeinen werden Befehle an einen bestimmten Empfänger gesendet, und ein Ergebnis wird an den Client zurückgeschickt.

Nehmen wir die einfache HTTP-Anwendung, die wir in Kapitel 3 verwendet haben. Du hast eine einfache HTTP-Anfrage gesendet. Wie wir schon gesagt haben, war das ein Befehl. Die Anwendung empfängt diesen Befehl, verarbeitet ihn und liefert ein Ergebnis.

Veranstaltungen

Ereignisse sind Aktionen, die erfolgreich abgeschlossen wurden. Ein Ereignis stellt eine Tatsache dar, etwas, das passiert ist: ein Tastendruck, ein Fehler, ein Befehl, irgendetwas, das für die jeweilige Organisation oder das System wichtig ist. Ein Ereignis kann das Ergebnis der Arbeit eines Befehls sein.

Gehen wir noch einmal zurück zum Beispiel der HTTP-Anfrage. Sobald die Antwort geschrieben wurde, wird sie zu einem Ereignis. Wir haben eine HTTP-Anfrage und ihre Antwort gesehen. Dieses Ereignis kann in ein Protokoll geschrieben oder an interessierte Parteien gesendet werden, damit sie wissen, was passiert ist.

Ereignisse sind unveränderlich. Du kannst ein Ereignis nicht löschen. Allerdings kannst du die Vergangenheit nicht ändern. Wenn du eine zuvor gesendete Tatsache widerlegen willst, musst du ein anderes Ereignis abfeuern, das die Tatsache ungültig macht. Die mitgeführten Tatsachen werden erst durch eine andere Tatsache, die das aktuelle Wissen begründet, irrelevant.

Nachrichten

Aber wie veröffentlicht man diese Ereignisse? Es gibt viele Möglichkeiten. Heutzutage sind Lösungen wie Apache Kafka oder Apache ActiveMQ (beide werden in Kapitel 11 behandelt) sehr beliebt. Sie fungieren als Broker zwischen Produzenten und Konsumenten. Im Wesentlichen werden unsere Ereignisse in Topics oder Warteschlangen geschrieben. Um diese Ereignisse zu schreiben, sendet die Anwendung eine Nachricht an den Broker, die ein bestimmtes Ziel (die Warteschlange oder das Topic) ansteuert.

Eine Nachricht ist eine in sich geschlossene Datenstruktur, die das Ereignis und alle relevanten Details über das Ereignis beschreibt, z. B. wer es ausgelöst hat, zu welchem Zeitpunkt es ausgelöst wurde und möglicherweise seine eindeutige ID. Im Allgemeinen ist es besser, das Ereignis selbst geschäftsorientiert zu halten und zusätzliche Metadaten für die technischen Aspekte zu verwenden.

Um Ereignisse zu konsumieren, abonnierst du die Warteschlange oder das Thema mit den Ereignissen, an denen du interessiert bist, und empfängst die Nachrichten. Du packst das Ereignis aus und kannst auch die zugehörigen Metadaten abrufen (z. B. wann das Ereignis eingetreten ist, wo es eingetreten ist usw.). Die Verarbeitung eines Ereignisses kann zur Veröffentlichung anderer Ereignisse (wiederum in Nachrichten verpackt und an ein bekanntes Ziel gesendet) oder zur Ausführung von Befehlen führen.

Makler und Nachrichten können auch Befehle übermitteln. In diesem Fall enthält die Nachricht die Beschreibung der auszuführenden Aktion, und eine weitere Nachricht (möglicherweise mehrere Nachrichten) würde bei Bedarf das Ergebnis übermitteln.

Befehle statt Ereignisse: Ein Beispiel

Schauen wir uns ein Beispiel an, um die Unterschiede zwischen Befehlen und Ereignissen zu verdeutlichen. Stell dir einen E-Commerce-Shop vor, wie er in Abbildung 4-2 dargestellt ist. Der Nutzer wählt eine Reihe von Produkten aus und schließt die Bestellung ab (Zahlungsvorgang, Liefertermin usw.).

Abbildung 4-2. Vereinfachte Architektur eines E-Commerce-Shops

Der Nutzer sendet einen Befehl (z. B. über eine HTTP-Anfrage) an den Shop-Dienst mit den Artikeln, die der Nutzer erhalten möchte. In einer herkömmlichen Anwendung würde ShopService, sobald es den Befehl erhält, OrderService aufrufen und eine Methode order mit dem Benutzernamen, der Liste der Artikel (Warenkorb) usw. aufrufen. Der Aufruf der Methode order ist ein Befehl. Das macht ShopService von OrderService abhängig und reduziert die Autonomie der Komponenten: ShopService kann nicht ohne OrderService arbeiten. Wir schaffen einen verteilten Monolithen, eine verteilte Anwendung, die zusammenbricht, sobald einer ihrer Teile fehlschlägt.¹

Sehen wir uns den Unterschied an, wenn wir anstelle eines Befehls zwischen ShopService und OrderService ein Ereignis veröffentlichen. Sobald der Nutzer die Bestellung abgeschlossen hat, sendet die Anwendung immer noch einen Befehl an ShopService. Dieses Mal wandelt ShopService diesen Befehl jedoch in ein Ereignis um: Eine neue Bestellung wurde aufgegeben. Das Ereignis enthält den Nutzer, den Warenkorb usw. Das Ereignis ist ein Fakt, der in ein Protokoll geschrieben oder in eine Nachricht verpackt und an einen Broker gesendet wird.

Auf der anderen Seite beobachtet OrderService das Ereignis " Eine neue Bestellung wurde aufgegeben", indem es liest, wo diese Ereignisse gespeichert sind. Wenn ShopService das Ereignis sendet, empfängt es es und kann es verarbeiten.

Bei dieser Architektur ist ShopService nicht von OrderService abhängig. Außerdem ist OrderService nicht von ShopService abhängig und würde jedes beobachtete Ereignis verarbeiten, unabhängig vom Absender. Eine mobile Anwendung kann zum Beispiel das gleiche Ereignis senden, wenn der Nutzer eine Bestellung von einem Mobiltelefon aus validiert.

Mehrere Komponenten können Ereignisse konsumieren(Abbildung 4-3). Zum Beispiel behält StatisticsService zusätzlich zu OrderService den Überblick über die am meisten bestellten Artikel. Sie konsumiert dasselbe Ereignis, ohne dass ShopService geändert werden muss, um sie zu empfangen.

Eine Komponente, die Ereignisse beobachtet, kann daraus neue Ereignisse ableiten, z.B,StatisticsService könnte zum Beispiel die Bestellung analysieren und Empfehlungen berechnen. Diese Empfehlungen könnten als ein weiteres Faktum betrachtet und somit als Ereignis kommuniziert werden.ShopService könnte diese Ereignisse beobachten und verarbeiten, um die Artikelauswahl zu beeinflussen. StatisticsService und ShopService sind jedoch unabhängig voneinander. Das Wissen ist kumulativ und entsteht durch den Empfang neuer Ereignisse und die Ableitung neuer Fakten aus den empfangenen Ereignissen, wie es StatisticsService tut.

Wie in Abbildung 4-3 dargestellt, können wir Nachrichtenwarteschlangen verwenden, um unsere Ereignisse zu transportieren. Diese Ereignisse werden in Nachrichten verpackt und an bekannte Ziele (orders undrecommendations).OrderService und StatisticsService konsumieren und verarbeiten die Nachrichten unabhängig voneinander.

Abbildung 4-3. Architektur des E-Commerce-Shops mit Ereignissen und Nachrichtenwarteschlangen

Für diese Ziele ist es wichtig, dass die Ereignisse in einer geordneten Reihenfolge aufbewahrt werden. Indem diese Reihenfolge beibehalten wird, kann das System in der Zeit zurückgehen und die Ereignisse erneut verarbeiten.Ein solcher Wiedergabemechanismus, der in der Kafka-Welt sehr beliebt ist, hat mehrere Vorteile. Nach einer Katastrophe kann man mit einem sauberen Zustand neu starten, indem alle gespeicherten Ereignisse erneut verarbeitet werden. Wenn wir dann zum Beispiel den Empfehlungsalgorithmus der Statistikdienste ändern, kann er das gesamte Wissen neu sammeln und neue Empfehlungen ableiten.

Auch wenn die Ereignisemission in diesem Beispiel eindeutig klingt, ist das nicht immer der Fall. Ereignisse können zum Beispiel durch Schreibvorgänge in der Datenbank erzeugt werden.²

Unabhängigkeit

Die Verwendung von Nachrichten erhöht die Unabhängigkeit der Komponenten, die das System bilden. Diese Unabhängigkeit ist relativ. Während die Komponenten einander nicht kennen, müssen die ausgesendeten Ereignisse und die Empfänger ein gemeinsames Vokabular verwenden. Dieses Vokabular ist oft geschäftsbezogen, kann aber auch technisch sein.

Im Fall von Geschäftsereignissen ergeben sie sich natürlich aus der Analyse der Domäne. Wenn du dein System in Form von Befehlen und Ereignissen betrachtest, konzentrierst du dich auf das Verhalten und nicht auf die Struktur (wir konzentrieren uns meistens auf die Struktur). Wenn du dich auf das Verhalten konzentrierst, liegt das Hauptaugenmerk auf der Kommunikation und den Arbeitsabläufen und du kannst bestimmen, wer für was verantwortlich ist und welche Art von Interaktionen (Befehle oder Ereignisse) am besten geeignet sind. Aus dieser Analyse kannst du die Art der Interaktionen, die Daten, die sie kapseln müssen, und den Ablauf in deinem System bestimmen.

Workshops und Techniken - wie Event Storming und die auf der Seite der Domain-Driven Design Crewaufgeführten - könnendiese Analyse erleichtern. Mit diesen Techniken kannst du die Ereignisse, Befehle und Komponenten extrahieren und vor allem die Kommunikation hervorheben.

Befehle und Ereignisse bilden die Grundlage für die meisten Interaktionen. Obwohl wir hauptsächlich Befehle verwenden, bieten Ereignisse erhebliche Vorteile. Ereignisse sind Fakten. Ereignisse erzählen eine Geschichte, die Geschichte deines Systems, eine Erzählung, die die Entwicklung deines Systems beschreibt. In reaktiven Systemen werden Ereignisse in Nachrichten verpackt und diese Nachrichten werden an Ziele gesendet, die von Message Brokern wie AMQP oder Kafka transportiert werden(Abbildung 4-4). Ein solcher Ansatz löst zwei wichtige architektonische Probleme, die sich aus verteilten Systemen ergeben. Erstens geht er auf natürliche Weise mit der Asynchronität der realen Welt um. Zweitens bindet er Dienste zusammen, ohne sich auf eine starke Kopplung zu verlassen. An den Kanten des Systems verwendet dieser Ansatz die meiste Zeit Befehle und verlässt sich oft auf HTTP.

Abbildung 4-4. Überblick über ein reaktives System

Dieser asynchrone Message-Passing-Aspekt reaktiver Systeme bildet das Bindegewebe. Er verleiht den Anwendungen, die das System bilden, nicht nur mehr Autonomie und Unabhängigkeit, sondern ermöglicht auch Widerstandsfähigkeit und Elastizität. Du fragst dich vielleicht, wie das geht, und bekommst im nächsten Abschnitt den Anfang unserer Antwort.

Zielorte und räumliche Entkopplung

Die reaktiven Anwendungen, die ein reaktives System bilden, kommunizieren über Nachrichten. Sie abonnieren Ziele und empfangen die Nachrichten, die von anderen Komponenten an diese Ziele gesendet werden. Diese Nachrichten können Befehle oder Ereignisse enthalten, wobei, wie im vorherigen Abschnitt beschrieben, Ereignisse interessante Vorteile bieten. Diese Ziele sind nicht an bestimmte Komponenten oder Instanzen gebunden. Sie sind virtuell. Die Komponenten müssen nur den Namen des Ziels kennen (in der Regel geschäftsbezogen, z. B. orders), nicht aber, wer es produziert oder konsumiert. Das ermöglicht Standorttransparenz.

Standort-Transparenz

Die Standorttransparenz bezieht sich auf die Verwendung von Namen zur Identifizierung von Ressourcen und nicht auf ihren tatsächlichen Standort. Zum Beispiel würdest du https://www.oreilly.com/ verwenden und nicht die 23.43.200.127 IP-Adresse. Der Hauptvorteil der Standorttransparenz ist, dass es keine Rolle mehr spielt, wo sich die Ressource befindet. In reaktiven Systemen sorgen Zielorte für Standorttransparenz. Reaktive Systeme verwenden diese identifizierbaren Zielorte anstelle des genauen Standorts des Verbrauchers.

Wenn du Kubernetes verwendest, kannst du davon ausgehen, dass die Standorttransparenz bereits für dich verwaltet wird. Tatsächlich kannst du Kubernetes-Dienste nutzen, um die Standorttransparenz zu implementieren. Du hast einen einzigen Endpunkt, der an eine Gruppe ausgewählter Pods delegiert. Diese Standorttransparenz ist jedoch etwas eingeschränkt und oft an HTTP- oder Anfrage/Antwort-Protokolle gebunden. Andere Umgebungen können eine Service-Discovery-Infrastruktur wie HashiCorp Consul oder Netflix Eureka nutzen.

Durch die Verwendung von Nachrichten, die an ein Ziel gesendet werden, kannst du als Absender ignorieren, wer genau die Nachricht erhalten wird. Du weißt nicht, ob jemand gerade verfügbar ist oder ob mehrere Komponenten oder Instanzen auf deine Nachricht warten. Die Anzahl der Konsumenten kann sich zur Laufzeit ändern; es können weitere Instanzen erstellt, verschoben oder zerstört und neue Komponenten eingesetzt werden. Aber für dich als Absender ist es nicht wichtig, das zu wissen. Du verwendest einfach ein bestimmtes Ziel. Veranschaulichen wir uns die Vorteile dieser Adressierbarkeit anhand des Beispiels aus dem vorherigen Abschnitt.ShopService sendet order placed Ereignisse, die in Nachrichten an das Ziel orders übertragen werden(Abbildung 4-3). Es ist möglich, dass während einer ruhigen Zeit nur eine einzige Instanz von OrderService läuft. Wenn es nicht viele Bestellungen gibt, warum sollte man sich die Mühe machen, mehr davon zu haben? Wir könnten uns sogar vorstellen, keine Instanz zu haben und eine zu instanziieren, wenn wir eine Bestellung erhalten. Serverlose Plattformen bieten diese Möglichkeit der Skalierung von Null an. Mit der Zeit bekommt dein Laden jedoch mehr Kunden, und eine einzige Instanz reicht vielleicht nicht mehr aus. Dank derStandorttransparenz können wir andere Instanzen von OrderService starten, um die Last zu teilen(Abbildung 4-5).ShopService wird nicht verändert und ignoriert diese neue Topologie.

Abbildung 4-5. Elastizität durch die Verwendung von Message Passing

Die Art und Weise, wie die Last auf die Konsumenten verteilt wird, ist für den Absender ebenfalls irrelevant. Es kann ein Round-Robin-Verfahren, eine lastbasierte Auswahl oder etwas Ausgeklügelteres sein. Wenn sich die Last wieder normalisiert, kann das System die Anzahl der Instanzen reduzieren und Ressourcen sparen. Diese Art von Elastizität funktioniert perfekt für zustandslose Dienste. Bei zustandsabhängigen Diensten kann es schwieriger sein, da die Instanzen den Zustand gemeinsam nutzen müssen. Es gibt jedoch Lösungen (wenn auch nicht ohne Einschränkungen), wie Kubernetes StatefulSet oder ein In-Memory-Datengitter, um den Zustand zwischen den Instanzen desselben Dienstes zu koordinieren. Message Passing ermöglicht auch eine Replikation. Nach dem gleichen Prinzip können wir die aktive Instanz von OrderService schatten und übernehmen, wenn die primäre Instanz fehlschlägt(Abbildung 4-6). So wird eine Unterbrechung des Dienstes vermieden. Diese Art von Failover kann auch eine gemeinsame Nutzung des Status erfordern.

Abbildung 4-6. Ausfallsicherheit durch die Verwendung von Message Passing

Durch den Einsatz von Message Passing wird unser System nicht nur asynchron, sondern auch elastisch und widerstandsfähig. Wenn du dein System entwirfst, listest du die Ziele auf, die das gewünschte Kommunikationsmuster implementieren. Im Allgemeinen würdest du ein Ziel pro Ereignistyp verwenden, aber das ist nicht unbedingt der Fall. Vermeide es jedoch unbedingt, ein Ziel pro Komponenteninstanz zu verwenden. Das führt zu einer Kopplung zwischen Sender und Empfänger, was die Vorteile zunichte macht. Außerdem verringert es die Erweiterbarkeit. Schließlich ist es wichtig, die Menge der Ziele stabil zu halten. Das Ändern eines Ziels würde die Komponenten, die es verwenden, zerstören oder dich dazu zwingen, Umleitungen einzurichten.

Zeitliche Entkopplung

Standorttransparenz ist nicht der einzige Vorteil: Die asynchrone Weitergabe von Nachrichten ermöglicht auch eine zeitliche Entkopplung.

Moderne Nachrichten-Backbones wie AMQP 1.0, Apache Kafka und sogar Java Message Service (JMS) ermöglichen eine zeitliche Entkopplung. Bei diesen Event-Brokern gehen Ereignisse nicht verloren, wenn es keine Konsumenten gibt. Die Ereignisse werden gespeichert und später zugestellt. Jeder Broker hat seine eigene Methode. AMQP 1.0 verwendet beispielsweise persistente Nachrichten und dauerhafte Abonnenten, um die Zustellung von Nachrichten zu gewährleisten. Kafka speichert Datensätze in einem dauerhaften, fehlertoleranten, geordneten Protokoll. Die Datensätze können so lange abgerufen werden, wie sie im Topic gespeichert bleiben.

Wenn unser ShopService die abgeschlossenen Aufträge als Ereignisse sendet, muss es nicht wissen, ob OrderService verfügbar ist. Es weiß, dass sie irgendwann verarbeitet werden. Wenn zum Beispiel keine Instanzen von OrderService verfügbar sind, wenn ShopService das Ereignis sendet, ist es nicht verloren. Wenn eine Instanz bereit ist, erhält sie die ausstehenden Aufträge und verarbeitet sie. Der Benutzer wird dann asynchron mit einer E-Mail benachrichtigt.

Natürlich muss der Message Broker verfügbar und erreichbar sein. Die meisten Message Broker verfügen über Replikationsfunktionen, die Probleme bei Nichtverfügbarkeit und Nachrichtenverluste verhindern.

Hinweis

Es wird immer üblicher, Ereignisse in einem Ereignisprotokoll zu speichern. Eine solche geordnete und nur anhängende Struktur stellt die gesamte Geschichte deines Systems dar. Jedes Mal, wenn sich der Zustand ändert, hängt das System den neuen Zustand an das Protokoll an.

Die zeitliche Entkopplung erhöht die Unabhängigkeit unserer Komponenten. Zusammen mit anderen Funktionen, die durch die asynchrone Nachrichtenübermittlung ermöglicht werden, erreicht die zeitliche Entkopplung ein hohes Maß an Unabhängigkeit zwischen unseren Komponenten und reduziert die Kopplung auf einMinimum.

Die Rolle der nicht blockierenden Ein-/Ausgabe

An dieser Stelle fragst du dich vielleicht, was der Unterschied zwischen einer Anwendung, die Kafka oder AMQP nutzt, und einem reaktiven System ist. Message Passing ist das Wesen reaktiver Systeme, und die meisten von ihnen basieren auf einer Art Message Broker. Message Passing ermöglicht Ausfallsicherheit und Elastizität, was zu Reaktionsfähigkeit führt. Es fördert die räumliche und zeitliche Entkopplung und macht unser System viel robuster.

Aber reaktive Systeme tauschen nicht nur Nachrichten aus. Das Senden und Empfangen von Nachrichten muss effizient erfolgen. Um dies zu erreichen, fördert Reactive die Verwendung von nicht blockierenden I/Os.

Blockierende Netzwerk-E/A, Threads und Gleichzeitigkeit

Um die Vorteile von nicht blockierenden E/As zu verstehen, müssen wir wissen, wie blockierende E/As funktionieren. Nehmen wir zur Veranschaulichung eine Client/Server-Interaktion. Wenn ein Client eine Anfrage an einen Server sendet, verarbeitet der Server diese und sendet eine Antwort zurück. HTTP zum Beispiel folgt diesem Prinzip. Damit dies geschehen kann, müssen sowohl der Client als auch der Server eine Verbindung aufbauen, bevor die Interaktion beginnt. Wir werden nicht in die Tiefen des Sieben-Schichten-Modells und des Protokollstapels gehen, die an dieser Interaktion beteiligt sind; du kannst viele Artikel zu diesem Thema im Internet finden.

Hinweis

Die Beispiele aus diesem Abschnitt können direkt in deiner IDE ausgeführt werden. Verwende chapter-4/non-blocking-io/src/main/java/org/acme/client/EchoClient.java, um den gestarteten Server aufzurufen. Achte darauf, dass nicht mehrere Server gleichzeitig laufen, da sie alle denselben Port (9999) verwenden.

Um diese Verbindung zwischen dem Client und dem Server herzustellen, verwenden wir sockets, wie in Beispiel 4-1 gezeigt.

Beispiel 4-1. Ein Single-Thread-Echoserver mit blockierender E/A(chapter-4/non-blocking-io/src/main/java/org/acme/blocking/BlockingEchoServer.java)

int port = 9999;

// Create a server socket
try (ServerSocket server = new ServerSocket(port)) {
    while (true) {

        // Wait for the next connection from a client
        Socket client = server.accept();

        PrintWriter response = new PrintWriter(client.getOutputStream(), true);
        BufferedReader request = new BufferedReader(
                new InputStreamReader(client.getInputStream()));

        String line;
        while ((line = request.readLine()) != null) {
            System.out.println("Server received message from client: " + line);
            // Echo the request
            response.println(line);

            // Add a way to stop the application.
            if ("done".equalsIgnoreCase(line)) {
                break;
            }
        }
        client.close();
    }
}

Der Client und der Server müssen sich an einen Socket binden, der die Verbindung herstellt. Der Server wartet an seinem Socket darauf, dass der Client eine Verbindung herstellt. Sobald die Verbindung hergestellt ist, können der Client und der Server Daten aus dem Socket, der an diese Verbindung gebunden ist, lesen und schreiben.

Traditionell werden Anwendungen, weil es einfacher ist, mit einem synchronen Entwicklungsmodell entwickelt. Ein solches Entwicklungsmodell führt Anweisungen sequentiell, also nacheinander, aus. Wenn solche Anwendungen über das Netzwerk interagieren, wird erwartet, dass sie auch für E/A ein synchrones Entwicklungsmodell verwenden. Dieses Modell verwendet synchrone Kommunikation und blockiert die Ausführung, bis die Operation abgeschlossen ist. In Beispiel 4-1 warten wir auf eine Verbindung und behandeln sie synchron. Wir lesen und schreiben mit synchronen APIs. Das ist einfacher, führt aber zur Verwendung von blockierender E/A.

Bei der blockierenden E/A wird, wenn der Client eine Anfrage an den Server sendet, der Socket, der diese Verbindung verarbeitet, und der entsprechende Thread, der von ihm liest, blockiert, bis einige Lesedaten erscheinen. Die Bytes werden im Netzwerkpuffer angesammelt, bis alles gelesen und zur Verarbeitung bereit ist. Bis der Vorgang abgeschlossen ist, kann der Server nichts weiter tun als warten.

Die Konsequenz dieses Modells ist, dass wir nicht mehr als eine Verbindung in einem einzigen Thread bedienen können. Wenn der Server eine Verbindung erhält, verwendet er diesen Thread, um die Anfrage zu lesen, zu verarbeiten und die Antwort zu schreiben. Dieser Thread ist blockiert, bis das letzte Byte der Antwort auf die Leitung geschrieben wurde. Eine einzige Client-Verbindung blockiert den Server! Nicht sehr effizient, oder?

Um mit diesem Ansatz gleichzeitige Anfragen zu bearbeiten, gibt es nur die Möglichkeit, mehrere Threads zu verwenden. Wir müssen für jede Client-Verbindung einen neuen Thread zuweisen. Um mehr Clients zu bearbeiten, musst du mehr Threads verwenden und jede Anfrage in einem anderen Worker-Thread bearbeiten; siehe Beispiel 4-2.

Beispiel 4-2. Prinzipien eines Multithreading-Servers mit blockierender E/A

while (listening) {
    accept a connection;
    create a worker thread to process the client request;
}

Um dieses Prinzip umzusetzen, brauchen wir einen Thread-Pool(Worker-Pool). Wenn sich der Client verbindet, nehmen wir die Verbindung an und verlagern die Verarbeitung auf einen separaten Thread. So kann der Server-Thread weiterhin andere Verbindungen annehmen, wie in Beispiel 4-3 gezeigt.

Beispiel 4-3. Ein Multithreading-Echoserver mit blockierender E/A(chapter-4/non-blocking-io/src/main/java/org/acme/blocking/BlockingWithWorkerEchoServer.java)

int port = 9999;
ExecutorService executors = Executors.newFixedThreadPool(10); 

// Create a server socket
try (ServerSocket server = new ServerSocket(port)) {
    while (true) {

        // Wait for the next connection from a client
        Socket client = server.accept();

        executors.submit(() -> {                                    
            try {
                PrintWriter response =
                new PrintWriter(client.getOutputStream(), true);
                BufferedReader request = new BufferedReader(
                        new InputStreamReader(client.getInputStream()));

                String line;
                while ((line = request.readLine()) != null) {
                    System.out.println(Thread.currentThread().getName() +
                            " - Server received message from client: " + line);
                    // Echo the request
                    response.println(line);

                    // Add a way to stop the application.
                    if ("done".equalsIgnoreCase(line)) {
                        break;
                    }
                }
                client.close();
            } catch (Exception e) {
                System.err.println("Couldn't serve I/O: " + e.toString());

            }
        });
    }
}

Erstelle einen Worker-Thread-Pool, um die Anfrage zu bearbeiten.

Übertrage die Bearbeitung der Anfrage auf einen Thread aus dem Thread-Pool. Der Rest des Codes bleibt unverändert.

Das ist das Modell, das standardmäßig in traditionellen Java-Frameworks wie Jakarta EE oder Spring verwendet wird.³ Auch wenn diese Frameworks unter der Haube nicht blockierende E/A verwenden, nutzen sie Worker-Threads, um die Anfragen zu bearbeiten. Dieser Ansatz hat jedoch viele Nachteile, darunter:

• Mit zunehmender Anzahl von Verbindungen verbraucht das Erzeugen mehrerer Threads und das Wechseln zwischen ihnen nicht nur Speicher, sondern auch CPU-Zyklen.

• Zu jedem beliebigen Zeitpunkt könnten mehrere Threads auf die Client-Anfragen warten. Das ist eine enorme Ressourcenverschwendung.

• Deine Gleichzeitigkeit (die Anzahl der Anfragen, die du zu einem bestimmten Zeitpunkt bearbeiten kannst - im vorherigen Beispiel 10) wird durch die Anzahl der Threads begrenzt, die du erstellen kannst.

In öffentlichen Clouds treibt der Blocking I/O-Ansatz die monatliche Rechnung in die Höhe, in privaten Clouds verringert er die Bereitstellungsdichte. Daher ist dieser Ansatz nicht ideal, wenn du viele Verbindungen verarbeiten oder Anwendungen implementieren musst, die mit viel I/O zu tun haben. Zum Glück gibt es eine Alternative.

Wie funktioniert Nonblocking I/O?

Die Alternative ist Nonblocking I/O. Der Unterschied ist schon im Namen zu erkennen. Anstatt auf den Abschluss der Übertragung zu warten, wird der Aufrufer nicht blockiert und kann seine Bearbeitung fortsetzen. Die Magie findet im Betriebssystem statt. Bei nonblocking I/O stellt das Betriebssystem die Anfragen in eine Warteschlange. Das System verarbeitet die eigentliche I/O in der Zukunft. Wenn die I/O abgeschlossen und die Antwort fertig ist, erfolgt eine Fortsetzung, die oft als Callback implementiert ist, und der Aufrufer erhält das Ergebnis.

Fortsetzungen

Der Begriff Continuation stammt aus der funktionalen Programmierung.Der Continuation-Passing-Stil(CPS) ist ein Programmierstil, bei dem die Kontrolle explizit in Form einer Continuation weitergegeben wird. Er steht im Gegensatz zum imperativen Stil, dem üblichen Programmierstil, bei dem jeder Befehl der Reihe nach ausgeführt wird.

Um die Vorteile besser zu verstehen und zu sehen, wie diese Fortsetzungen funktionieren, müssen wir unter die Haubeschauen: Wie wird Non-Blocking I/O implementiert? Wir haben bereits eine Warteschlange erwähnt. Das System reiht I/O-Operationen in eine Warteschlange ein und kehrt sofort zurück, sodass der Aufrufer nicht blockiert wird, während er auf den Abschluss der I/O-Operationen wartet. Wenn eine Antwort zurückkommt, speichert das System das Ergebnis in einer Struktur. Wenn der Aufrufer das Ergebnis benötigt, fragt er das System ab, um zu sehen, ob die Operation abgeschlossen wurde(Beispiel 4-4).

Beispiel 4-4. Ein Echoserver, der nicht blockierende E/A verwendet(chapter-4/non-blocking-io/src/main/java/org/acme/nio/NonBlockingServer.java)

InetSocketAddress address = new InetSocketAddress("localhost", 9999);
Selector selector = Selector.open();
ServerSocketChannel channel = ServerSocketChannel.open();
channel.configureBlocking(false);

channel.socket().bind(address);
// Server socket supports only ACCEPT
channel.register(selector, SelectionKey.OP_ACCEPT);

while (true) {
    int available = selector.select(); // wait for events
    if (available == 0) {
        continue;  // Nothing ready yet.
    }

    // We have the request ready to be processed.
    Set<SelectionKey> keys = selector.selectedKeys();
    Iterator<SelectionKey> iterator = keys.iterator();
    while (iterator.hasNext()) {
        SelectionKey key = iterator.next();
        if (key.isAcceptable()) {
            // --  New connection --
            SocketChannel client = channel.accept();
            client.configureBlocking(false);
            client.register(selector, SelectionKey.OP_READ);
            System.out.println("Client connection accepted: "
                + client.getLocalAddress());
        } else if (key.isReadable()) {
            // --  A client sent data ready to be read and we can write --
            SocketChannel client = (SocketChannel) key.channel();
            // Read the data assuming the size is sufficient for reading.
            ByteBuffer payload = ByteBuffer.allocate(256);
            int size = client.read(payload);
            if (size == -1 ) { // Handle disconnection
                System.out.println("Disconnection from "
                    + client.getRemoteAddress());
                channel.close();
                key.cancel();
            } else {
                String result = new String(payload.array(),
                    StandardCharsets.UTF_8).trim();
                System.out.println("Received message: " + result);
                if (result.equals("done")) {
                    client.close();
                }
                payload.rewind(); // Echo
                client.write(payload);
            }
        }
        // Be sure not to handle it twice.
        iterator.remove();
    }
}

Nonblocking I/O führt ein paar neue Konzepte ein:

• Wir verwenden nicht InputStream oder OutputStream (die von Natur aus blockierend sind), sondern Buffer, das eine temporäre Speicherung ist.

• Channel kann als Endpunkt für eine offene Verbindung angesehen werden.

• Selector ist der Eckpfeiler der blockierungsfreien E/A in Java.

Selector verwaltet mehrere Kanäle, entweder Server- oder Client-Kanäle. Wenn du Non-Blocking I/O verwendest, erstellst du . Jedes Mal, wenn du mit einem neuen Kanal zu tun hast, registrierst du diesen Kanal im Selektor mit den Ereignissen, an denen du interessiert bist (akzeptieren, bereit zum Lesen, bereit zum Schreiben). Selector

Dann fragt dein Code Selector mit nur einem Thread ab, ob der Kanal bereit ist. Wenn der Kanal zum Lesen oder Schreiben bereit ist, kannst du mit dem Lesen und Schreiben beginnen. Wir brauchen nicht für jeden Kanal einen Thread, und ein einziger Thread kann viele Kanäle verwalten.

Der Selektor ist eine Abstraktion der Non-Blocking-I/O-Implementierung, die vom zugrunde liegenden Betriebssystem bereitgestellt wird. Je nach Betriebssystem gibt es verschiedene Ansätze.

Erstens wurde select in den 1980er Jahren implementiert. Es unterstützt die Registrierung von 1.024 Sockeln. In den 80er Jahren war das sicherlich genug, aber heute nicht mehr.

poll ist ein Ersatz für , das 1997 eingeführt wurde. Der wichtigste Unterschied ist, dass die Anzahl der Steckdosen nicht mehr begrenzt. Wie bei sagt dir das System jedoch nur, wie viele Kanäle bereit sind, nicht welche. Du musst über die Menge der Kanäle iterieren, um zu prüfen, welche bereit sind. Wenn es nur wenige Kanäle gibt, ist das kein großes Problem. Sobald die Anzahl der Kanäle mehr als hunderttausend beträgt, ist die Iterationszeit beträchtlich. select poll select

Dann erschien epoll im Jahr 2002 im Linux Kernel 2.5.44.Kqueue erschien im Jahr 2000 in FreeBSD und /dev/poll etwa zur gleichen Zeit in Solaris. Diese Mechanismen geben die Menge der Kanäle zurück, die bereit sind, verarbeitet zu werden - keine Iteration mehr über jeden Kanal! Schließlich bieten Windows-Systeme IOCP, eine optimierte Implementierung von select.

Wichtig ist, dass du dich daran erinnerst, dass du bei Non-Blocking I/O nur einen einzigen Thread brauchst, um mehrere Anfragen zu bearbeiten. Dieses Modell ist viel effizienter als Blocking I/O, da du keine Threads erstellen musst, um gleichzeitige Anfragen zu bearbeiten. Durch die Eliminierung dieser zusätzlichen Threads wird deine Anwendung viel effizienter, was den Speicherverbrauch angeht (etwa 1 MB pro Thread) und vermeidet die Verschwendung von CPU-Zyklen aufgrund von Kontextwechseln (1-2 Mikrosekunden pro Wechsel).⁴

Reaktive Systeme empfehlen den Einsatz von nonblocking I/O zum Empfangen und Senden von Nachrichten. So kann deine Anwendung mehr Nachrichten mit weniger Ressourcen verarbeiten. Ein weiterer Vorteil ist, dass eine Anwendung im Leerlauf so gut wie keinen Speicher oder CPUs verbraucht. Du musst keine Ressourcen im Voraus reservieren.

Reactor-Muster und Ereignisschleife

Nonblocking I/O gibt uns die Möglichkeit, mehrere gleichzeitige Anfragen oder Nachrichten mit einem einzigen Thread zu bearbeiten. Wie können wir diese gleichzeitigen Anfragen bearbeiten? Wie strukturieren wir unseren Code, wenn wir Nonblocking I/O verwenden? Die Beispiele im vorigen Abschnitt skalieren nicht gut; wir sehen schnell, dass die Implementierung einer REST-API mit einem solchen Modell ein Albtraum sein wird. Außerdem möchten wir die Verwendung von Worker-Threads vermeiden, da dies die Vorteile von Nonblocking I/O zunichte machen würde. Wir brauchen etwas anderes: das Reaktor-Muster.

Das Reaktor-Muster, das in Abbildung 4-7 dargestellt ist, ermöglicht es, E/A-Ereignisse mit Ereignishandlern zu verknüpfen. Der Reaktor, der Eckpfeiler dieses Mechanismus, ruft die Ereignishandler auf, wenn das erwartete Ereignis empfangen wird.

Der Zweck des Reaktor-Patterns ist es, die Erstellung eines Threads für jede Nachricht, Anfrage und Verbindung zu vermeiden. Dieses Pattern empfängt Ereignisse von mehreren Kanälen und verteilt sie sequentiell an die entsprechenden Event-Handler.

Abbildung 4-7. Das Reaktormuster

Die Implementierung des Reaktor-Musters verwendet eine Ereignisschleife(Abbildung 4-7). Es handelt sich dabei um einen Thread, der über die Menge der Kanäle iteriert, und wenn die Daten bereit sind, konsumiert zu werden, ruft die Ereignisschleife den zugehörigen Event-Handler sequentiell auf.

Wenn du Non-Blocking I/O und das Reactor-Pattern kombinierst, organisierst du deinen Code als eine Reihe von Event-Handlern. Dieser Ansatz funktioniert wunderbar mit reaktivem Code, da er das Konzept der Events, den Kern von Reactive, offenlegt.

Das Reaktormuster hat zwei Varianten:

• Das Multireactor-Pattern verwendet mehrere Ereignisschleifen (in der Regel eine oder zwei pro CPU-Kern), die die Gleichzeitigkeit der Anwendung erhöhen. Multireactor-Pattern-Implementierungen, wie z. B. Eclipse Vert.x, rufen die Event-Handler in einer Single-Thread-Methode auf, um Deadlock- oder State-Visibility-Probleme zu vermeiden.

• Das Proactor-Muster kann als eine asynchrone Version des Reactor-Musters angesehen werden. Lang laufende Event-Handler rufen nach ihrer Beendigung eine Fortsetzung auf. Solche Mechanismen ermöglichen es, nicht blockierende und blockierende E/A zu mischen(Abbildung 4-8).

Abbildung 4-8. Das Proactor-Muster

Du kannst sowohl blockierende als auch nicht blockierende Event-Handler einbinden, indem du ihre Ausführung auf separate Threads auslagerst, wenn dies unvermeidlich ist. Wenn ihre Ausführung abgeschlossen ist, ruft das Proactor-Muster die Fortsetzung auf. Wie du in Kapitel 6 sehen wirst, ist dies das Muster, das von Quarkus verwendet wird.

Anatomie der reaktiven Anwendungen

In den letzten Jahren sind viele Frameworks aufgetaucht, die reaktive Anwendungen unterstützen. Ihr Ziel ist es, die Implementierung reaktiver Anwendungen zu vereinfachen. Das erreichen sie, indem sie Primitive und APIs auf höherer Ebene bereitstellen, um Ereignisse und abstrakte nichtblockierende E/A zu behandeln.

Wie du vielleicht schon gemerkt hast, ist die Verwendung von nonblocking I/O nicht so einfach. Die Kombination mit einem Reactor-Pattern (oder einer Variante) kann sehr kompliziert sein. Zum Glück gibt es Frameworks, Bibliotheken und Toolkits, die diese Aufgabe übernehmen.Netty ist ein asynchrones, ereignisgesteuertes Netzwerkanwendungs-Framework, das nonblocking I/O nutzt, um hochgradig nebenläufige Anwendungen zu erstellen. Es ist die am häufigsten verwendete Bibliothek für nonblocking I/O in der Java-Welt. Aber Netty kann eine Herausforderung sein.Beispiel 4-5 implementiert den Echo-TCP-Server mit Netty.

Beispiel 4-5. Ein Echoserver mit Netty(chapter-4/non-blocking-io/src/main/java/org/acme/netty/NettyEchoServer.java)

public static void main(String[] args) throws Exception {
    new NettyServer(9999).run();
}

private final int port;

public NettyServer(int port) {
    this.port = port;
}

public void run() throws Exception {
    // NioEventLoopGroup is a multithreaded event loop that handles I/O operation.
    // The first one, often called 'boss', accepts an incoming connection.
    // The second one, often called 'worker', handles the traffic of the accepted
    // connection once the boss accepts the connection and registers the
    // accepted connection to the worker.
    EventLoopGroup bossGroup = new NioEventLoopGroup();

    EventLoopGroup workerGroup = new NioEventLoopGroup();
    try {
        // ServerBootstrap is a helper class that sets up a server.
        ServerBootstrap b = new ServerBootstrap();
        b.group(bossGroup, workerGroup)
                // the NioServerSocketChannel class is used to instantiate a
                // new Channel to accept incoming connections.
                .channel(NioServerSocketChannel.class)
                .childHandler(new ChannelInitializer<SocketChannel>() {
                    // This handler is called for each accepted channel and
                    // allows customizing the processing. In this case, we
                    // just append the echo handler.
                    @Override
                    public void initChannel(SocketChannel ch) {
                        ch.pipeline().addLast(new EchoServerHandler());
                    }
                });

        // Bind and start to accept incoming connections.
        ChannelFuture f = b.bind(port).sync();

        // Wait until the server socket is closed.
        f.channel().closeFuture().sync();
    } finally {
        workerGroup.shutdownGracefully();
        bossGroup.shutdownGracefully();
    }
}

private static class EchoServerHandler extends ChannelInboundHandlerAdapter {

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        // Write the received object, and flush
        ctx.writeAndFlush(msg);
    }
}

Das Vert.x-Toolkit, das auf Netty aufbaut, bietet Funktionen auf höherer Ebene, um reaktive Anwendungen wie HTTP-Clients und -Server, Messaging-Clients usw. zu erstellen. Normalerweise sieht ein Echo-TCP-Server mit Vert.x wie in Beispiel 4-6 aus.

Beispiel 4-6. Ein Echoserver mit Vert.x(chapter-4/non-blocking-io/src/main/java/org/acme/vertx/VertxEchoServer.java)

Vertx vertx = Vertx.vertx();
// Create a TCP server
vertx.createNetServer()
        // Invoke the given function for each connection
        .connectHandler(socket -> {
            // Just write the content back
            socket.handler(buffer -> socket.write(buffer));
        })
        .listen(9999);

Die meisten Java-Frameworks, die Reactive-Funktionen anbieten, basieren auf Netty oder Vert.x. Wie in Abbildung 4-9 dargestellt, folgen sie alle demselben Bauplan.

Abbildung 4-9. Die gemeinsame Architektur von reaktiven Frameworks

In der Regel verwenden Frameworks Netty oder Vert.x. Diese Schicht kümmert sich um Client-Verbindungen, ausgehende Anfragen und das Schreiben von Antworten. Mit anderen Worten: Sie verwaltet den E/A-Teil. Meistens implementiert diese Schicht das Reactor-Pattern (oder eine Variante davon) und bietet so ein Ereignisschleifen-basiertes Modell.

In der zweiten Schicht befindet sich das reaktive Framework an sich. Die Aufgabe dieser Schicht ist es, einfach zu verwendende APIs auf hoher Ebene bereitzustellen. Du nutzt diese APIs, um deinen Anwendungscode zu schreiben. Anstatt mit nicht blockierenden E/A-Kanälen umgehen zu müssen, stellt diese Schicht High-Level-Objekte wie HTTP-Anfragen, Antworten, Kafka-Nachrichten usw. bereit. Viel einfacher!

In der obersten Schicht schließlich befindet sich deine Anwendung. Dank des reaktiven Frameworks muss sich dein Code nicht mit nicht-blockierenden E/A-Konzepten befassen. Er kann sich auf eingehende Ereignisse konzentrieren und diese verarbeiten. Dein Code ist lediglich eine Sammlung von Ereignishandlern. Er kann die vom reaktiven Framework bereitgestellten Funktionen nutzen, um mit anderen Diensten oder Middleware zu interagieren.

Aber die Sache hat einen Haken.Der Event-Handler deines Codes wird über den Event-Loop-Thread (einen I/O-Thread) aufgerufen. Wenn dein Code diesen Thread blockiert, können keine anderen gleichzeitigen Ereignisse verarbeitet werden. Das wäre eine Katastrophe in Bezug auf Reaktionsfähigkeit und Gleichzeitigkeit. Die Konsequenz einer solchen Architektur liegt auf der Hand: Dein Code muss nonblocking sein. Er darf niemals die I/O-Threads blockieren, da diese selten sind und zur Bearbeitung mehrerer gleichzeitiger Anfragen verwendet werden. Um dies zu erreichen, könntest du die Verarbeitung einiger Ereignisse auf einen Worker-Thread auslagern (mit dem Proactor-Pattern). Auch wenn dadurch einige der Vorteile der blockierungsfreien E/A verloren gehen, ist dies manchmal die vernünftigste Wahl(Abbildung 4-10). Dennoch sollten wir dies nicht missbrauchen, da dadurch die reaktiven Vorteile verloren gehen und die Anwendung langsam wird. Die mehrfachen Kontextwechsel, die für die Verarbeitung eines Ereignisses auf einem Worker-Thread erforderlich sind, gehen zu Lasten der Reaktionszeit.

Abbildung 4-10. Ausführen einiger Event-Handler auf Worker-Threads

Unsere Anwendungen aus Kapitel 2 und Kapitel 3 beruhen in der Regel auf einem solchen Mechanismus.

Eine andere Möglichkeit besteht darin, nur auf nicht blockierenden Code zu setzen und sich auf asynchrone APIs zu verlassen, die vom reaktiven Framework zur Verfügung gestellt werden. Diese APIs wären nicht blockierend, und wenn die Geschäftslogik E/A beinhaltet, verwendet sie nicht blockierende E/A. Jedes Mal, wenn ein Event-Handler eine asynchrone Operation ausführt, wird ein anderer Handler (die Fortsetzung) registriert, und wenn das erwartete Ereignis eintrifft, ruft die Event-Schleife diesen auf. Auf diese Weise wird die Verarbeitung in kleinere Handler aufgeteilt, die asynchron laufen. Dieses Modell ist das effizienteste und umfasst die Konzepte hinter Reactive vollständig.

Zusammenfassung

Bei reaktiven Systemen geht es darum, bessere verteilte Systeme zu bauen. Sie zielen nicht darauf ab, die Natur verteilter Systeme zu verstecken, sondern im Gegenteil, sie zu umarmen.

In diesem Kapitel hast du das Folgende gelernt:

• Die vier Säulen reaktiver Systeme (asynchrone Nachrichtenübermittlung, Elastizität, Ausfallsicherheit und Reaktionsfähigkeit)

• Wie asynchrone Nachrichtenübermittlung Elastizität und Ausfallsicherheit ermöglicht und die Autonomie jeder einzelnen Komponente erhöht

• Die Rolle von Befehlen und Ereignissen in einem verteilten System

• Wie nonblocking I/O die Ressourcenauslastung in reaktiven Anwendungen verbessert

Aber dieser letzte Punkt hat einen entscheidenden Nachteil, denn wir müssen nicht blockierenden Code schreiben. Was für ein Zufall! Im nächsten Kapitel geht es genau darum!

¹ "Don't Build a Distributed Monolith" von Ben Christensen ist ein interessanter Vortrag über verteilte Monolithen und warum du sie vermeiden solltest.

² Dieses Muster wird Change Data Capture genannt. Frameworks wie Debezium sind ein Schlüsselelement reaktiver Systeme bei der Verwendung von Datenbanken, da die Ereignisse ohne Auswirkungen auf den Anwendungscode ausgelöst werden.

³ Wir beziehen uns hier auf das traditionelle Spring Framework. Reactive Spring basiert auf nonblocking I/O.

⁴ "Measuring Context Switching and Memory Overheads for Linux Threads" von Eli Bendersky liefert interessante Daten über die Kosten von Threads unter Linux.