Reactive Programming mit Spring Webflux

In diesem Artikel geben wir einen Überblick über Reactive Programming, erläutern, welche Prinzipien diesem zugrunde liegen und wann ein Einsatz sinnvoll sein kann. Anschließend zeigen wir, wie mithilfe des Spring-Webflux-Projekts eine reaktive Anwendung erstellt wird und welche Herausforderungen dieser Ansatz mit sich bringt. Dazu portieren wir einen Controller von dem klassischen Spring MVC nach Spring Webflux.

Was ist Reactive Programming?

Grundsätzlich betrachtet Reactive Programming Datenströme und Veränderungen innerhalb dieser Ströme. Im Einsatz befindet sich der Ansatz bereits seit den 1970er Jahren. Seit ca. 2010 erfreut er sich v. a. durch JavaScript-Frameworks zunehmender Beliebtheit. Das Reactive Manifesto von 2014 ist ein wichtiger Meilenstein dieser Bewegung. Ziel dieses Manifests war es, einen Ansatz für die Softwareentwicklung zu propagieren, mit dem Softwaresysteme einerseits flexibel, lose gekoppelt, fehlertolerant und skalierbar werden und andererseits leichter entwickelt und geändert werden können. Dazu wurden die zentralen Eigenschaften einer solchen Reactive Architecture festgehalten:

• Responsive: Das System antwortet mit einer schnellen und konsistenten Antwortzeit, solange eine Antwort möglich ist. Eine zuverlässige obere Schranke an die Antwortzeit stellt eine gleichbleibende Servicequalität sicher, vereinfacht Fehlerbehandlung und schafft Benutzervertrauen.
• Resilient: Das System bleibt im Falle eines (Teil-)Ausfalls weiterhin responsiv. Dies kann durch Replikation, Isolation und Containment der einzelnen Komponenten des Systems erreicht werden. Die Wiederherstellung einer ausgefallenen Komponente wird an eine andere (externe) Komponente delegiert.
• Elastic: Das System bleibt unter variierender Arbeitslast responsiv, indem die für die Verarbeitung eingehender Requests verfügbare Ressourcen automatisch hoch und runter skaliert werden. Die Skalierung kann vorausschauend oder reaktiv durchgeführt werden, sie findet jedoch stets kosteneffizient statt.
• Message Driven: Die Komponenten des Systems kommunizieren über das Versenden von Nachrichten, die asynchron vom Empfänger verarbeitet werden. Das schafft eine Abgrenzung der Komponenten untereinander und ermöglicht lose Kopplung und Isolation. Durch Monitoring der im System befindlichen Message Queues kann Lastmanagement, Elastizität und Flusskontrolle implementiert und so der „Back-pressure“ von überlaufenden Queues vermieden werden. Um Ressourcen möglichst effizient auszulasten, ist jegliche Kommunikation „non-blocking“.

Warum Non-blocking IO?

Das Reactive Manifesto empfiehlt Non-blocking IO, da dadurch die zur Verfügung stehenden Ressourcen effizienter genutzt werden können. Um diese Behauptung einordnen zu können, möchten wir zunächst den Unterschied zwischen Blocking und Non-blocking IO erklären.

Diagramm 1: Das Sequenzdiagramm zeigt, wie ein Thread bei einem ausgehenden Call in einen Wartezustand versetzt wird (roter Bereich) und erst nach dem Warten zur Bearbeitung des Requests wieder aktiv wird (grüner Bereich).

In klassischen Server-Applikationen (z. B. basierend auf Spring MVC) wird jeder Request an die Applikation durch einen eigens zugewiesenen Thread bearbeitet. Daher wird für jeden eingehenden Request ein neuer Thread erzeugt oder ein existierender Thread aus einem Thread-Pool bereitgestellt. Sobald einer dieser Threads mit einer externen Komponente (z. B. Datenbank, anderer Service) kommuniziert, wird der Thread von der Laufzeitumgebung unterbrochen und in einen Wartemodus versetzt, bis die Kommunikation von der externen Komponente beantwortet wurde. Man spricht in diesem Fall von Blocking IO. In diesem blockierten Zustand bleibt der Thread, bis die Laufzeitumgebung (z. B. Betriebssystem, JVM) ihn aufweckt und die Verarbeitung des Requests fortgesetzt werden kann. Blockierte Threads verbrauchen keine CPU-Zyklen und verhindern damit nicht, dass andere Threads ausgeführt werden, sie belegen aber dennoch Systemressourcen, wie etwa den Hauptspeicher. Entsprechend ist die maximale Anzahl parallel existierender Threads und damit die Anzahl paralleler Requests stark begrenzt. Diagramm 1 zeigt das Prinzip nochmal als Sequenzdiagramm. Hier steht nur ein einziger Thread für die Verarbeitung zur Verfügung, so dass Requests streng sequenziell abgearbeitet werden müssen.

Diagramm 2: In diesem Sequenzdiagramm sieht man, dass der ausführende Thread niemals wartet, sondern bei einem ausgehenden Call mit der Bearbeitung eines andere Requests weiter ausgelastet wird. Jeder Request verbraucht die gleichen CPU-Zyklen wie zuvor, es können aber mehr Requests gleichzeitig bedient werden.

Im Vergleich zur Blocking IO wird bei der Non-blocking IO der Thread nicht von der Laufzeitumgebung unterbrochen und blockiert. Die Abarbeitung von Requests durch die Applikation wird bei diesem Modell typischerweise von einer Handvoll Threads durchgeführt, indem diese in einer Schleife (dem sog. Event Loop) über alle offenen Kommunikationskanäle mit externen Komponenten iterieren und diese nach einer empfangenen Antwort fragen. Eine solche Antwort wird hier typischerweise als das Eintreffen eines Events modelliert. Sobald eine Antwort verfügbar ist, wird diese von dem Event Loop Thread verarbeitet und anschließend mit den verbleibenden Kommunikationskanälen fortgefahren. Wie in Diagramm 2 zu sehen, erlaubt diese Art der Kommunikation, dass eingehende Requests parallel verarbeitet werden, selbst dann, wenn nur ein einziger Event Loop Thread zur Verfügung steht. Dort, wo im Blocking-Modell der Thread in den Wartemodus versetzt wurde, ist Raum, um mit der Bearbeitung weiterer Requests fortzufahren. Welchen Effekt das auf den Durchsatz einer Applikation haben kann, wird mit dem Benchmark in Diagramm 3 deutlich (Details zum Benchmark auf GitHub).

Diagramm 3: Im Liniendiagramm ist der Durchsatz in Request pro Sekunde gegenüber der Gesamtanzahl der Request aufgetragen. Jeder Request benötigt eine Sekunde für die Bearbeitung. Wenn 256 Threads diese Requests ausführen, sieht man bereits bei einer geringen Gesamtanzahl von Request einen deutlichen Vorteil von Non-blocking IO.

Hier konnte der Durchsatz der Applikation bei gleichen Hardware-Ressourcen um 25 % gesteigert werden. Dieses Beispiel zeigt, dass man bei der Nutzung von Non-blocking IO durchaus Effizienzvorteile haben kann. Wir möchten jedoch ganz klar herausstellen, dass diese Vorteile nicht zwingend bei jeder Applikation zum Tragen kommen. Sollte die Applikation sehr CPU-intensive Aufgaben bearbeiten oder ist Latenz evtl. wichtiger als Durchsatz und Responsiveness, so ist der Non-blocking-Ansatz höchstwahrscheinlich eher nachteilig. Zudem ist Non-blocking-Code typischerweise komplexer und schwerer zu verstehen und zu debuggen, besonders wenn das Paradigma neu für die beteiligten Entwickler ist. Wie immer gilt es hier, den Einzelfall genau zu betrachten.

Vermeidung von Back-pressure

Ein weiterer Begriff, der im Reaktiven Manifest Verwendung findet, ist Back-pressure. Dabei handelt es sich um eine Analogie aus der Strömungsdynamik: Strömt Flüssigkeit durch ein Rohr, so entsteht ein Widerstand oder eine Kraft, die dem Strom entgegensteht. Etwas Ähnliches kann bei dem Datenfluss durch ein Softwaresystem beobachtet werden. Hier entsteht der Widerstand typischerweise durch limitierte Systemressourcen (z. B. Rechenzeit), die den Datenfluss ausbremsen können. Durch dieses Ausbremsen ist schnell die Responsiveness des Systems gefährdet, weshalb Back-pressure im Reactive Programming möglichst vermieden werden sollte. Dazu können grundsätzlich drei verschiedene Strategien herangezogen werden:

• Daten verwerfen: Die einfachste Strategie, um Back-pressure zu vermeiden, ist, die sich aufstauenden Daten einfach zu verwerfen. Die Implementierung einer solchen Strategie ist in den meisten Fällen trivial, aber offensichtlich nicht immer möglich.
• Daten puffern: Kann der Datenfluss nicht ausreichend schnell verarbeitet werden, ist das Puffern der eintreffenden Daten eine naheliegende Lösung. Hier ist jedoch Vorsicht geboten. Die Benutzung unbegrenzt wachsender Puffer ist eine häufige Ursache für Systemabstürze und damit ein Risiko für die Responsiveness des Systems. Hier kann es Sinn ergeben, ab einer festgelegten Puffergröße Daten zu verwerfen, anstatt den Puffer weiter wachsen zu lassen.
• Datenfluss kontrollieren: Die weitaus beste Strategie ist die Kontrolle des Datenflusses. Werden Daten von einem Produzenten zu einem Konsumenten übertragen, so kann der Konsument die Fließgeschwindigkeit kontrollieren, indem er regelmäßig dem Produzenten mitteilt, dass weitere Daten gesendet werden dürfen. Dieser Kontrollmechanismus birgt zwar Komplexität und Overhead, vermeidet aber große Puffer und Datenverlust und sollte daher in den meisten Fällen die bevorzugte Variante darstellen. Leider ist die Regulierung des Produzenten nicht immer möglich, z. B. wenn es sich dabei um einen Benutzer des Systems handelt.

In den meisten gängigen Frameworks für Reactive Programming wird letztere Strategie für die Vermeidung von Back-pressure genutzt. Mit Project Reactor wird im folgenden Abschnitt eines dieser Frameworks näher beleuchtet.

Spring Webflux und Project Reactor

Möchte man eine reaktive Anwendung im Java-Ökosystem bauen, so kommt man an der Reactive-Streams-Spezifikation nicht vorbei. Diese Spezifikation definiert einen Abstraktionslayer für Reactive Programming, der inzwischen als Teil des JDK 9+ von jeder Java-Anwendung genutzt werden kann. Die Spezifikation entstand aus einem Zusammenschluss verschiedener Reactive- bzw. Stream-Processing-Frameworks, die sich auf eine gemeinsame Basis einigen wollten und heute die Reactive Streams API aus dem JDK implementieren. Die beiden bekanntesten Frameworks sind hier RxJava und Project Reactor, es gibt aber viele weitere, auch außerhalb des Java-Ökosystems.

Das grundlegende Konzept von Reactive Streams basiert auf der Annahme, dass wir in unserem System einen Publisher haben, der einen (evtl. unendlichen) Stream von Events erzeugt und einen Subscriber, der diese Events empfangen und weiterverarbeiten will. Dazu meldet sich der Subscriber zunächst beim Publisher an; dabei entsteht eine Subscription. Der Publisher bestätigt die Anmeldung des Subscribers und wartet darauf, dass der Subscriber weitere Events vom Publisher anfragt (oder die Subscription beendet). Der Publisher kann dann die angefragte Menge von Events dem Subscriber per Push zur Verfügung stellen. Sämtliche Kommunikation zwischen Publisher und Subscriber ist dabei asynchron und non-blocking, so wie es das Reactive Manifesto fordert. Zudem erlaubt es die Reactive-Streams-Spezifikation, dass Publisher und Subscriber sich im gleichen oder in verschiedenen Prozessen befinden und damit evtl. über ein Netzwerk kommunizieren. Dafür nötige Netzwerkprotokolle werden von der Reactive-Streams-Spezifikation ebenfalls abgedeckt.

Das Spring Framework hat sich dafür entschieden, sämtliche Features für Reactive Programming auf der Reactive-Streams-Implementierung von Projekt Reactor (genauer: Reactor Core) basieren zu lassen. Dadurch begegnet man im Sourcecode des Spring-Projektes immer öfter den Klassen Mono und Flux, welche die beiden meistgenutzten Publisher aus dem Reactor Core sind. Ein Mono emittiert entweder kein oder genau ein Event, während ein Flux beliebig viele Events emittiert. Darauf aufbauend kümmert sich das Spring-Webflux-Projekt um alle Belange der Entwicklung reaktiver Webapplikationen.

Was macht Spring Webflux?

Mit Spring Webflux können auf einfache Weise reaktive Webanwendungen gebaut werden, d. h. sämtliche HTTP Requests werden von der resultierenden Anwendung mit Non-blocking IO abgearbeitet. Es stellt damit das Gegenstück zu dem schon deutlich länger etablierten Spring MVC dar, das aus der klassischen Blocking-IO-Welt stammt. Inzwischen werden zwar auch von Spring MVC einige Non-blocking-Operationen unterstützt, Webflux bietet darüber hinaus aber noch weitere Möglichkeiten. So bringt es unter anderem einen eigenen reaktiven HTTP Client mit, der für Non-blocking Requests gegen andere Dienste genutzt werden kann. Zudem bietet es eine Integration für einige populäre Non-blocking-Webserver, wie Netty oder Undertow.

Wie funktioniert das Zusammenspiel von Reactor und Webflux?

Um zu verstehen, wie sich Spring Webflux mit Project Reactor integriert, möchten wir uns eine beispielhafte Controller-Implementierung anschauen (gesamter Sourcecode verfügbar auf GitHub). Dazu implementieren wir diesen zunächst mit Spring MVC, also einem klassisch blockierenden Ansatz, da eine solche Implementierung den meisten Lesern geläufig sein dürfte:

1@RestController
2class CustomerController {
3
4   private final CustomerRepository repository;
5
6   CustomerController(final CustomerRepository repository) {
7      this.repository = repository;
8   }
9
10   @PostMapping("/customer")
11   Customer createNewCustomer(@RequestBody CustomerCreationRequest request) {
12      return repository.save(new Customer(request.firstName(), request.lastName()));
13   }
14}
15
16record CustomerCreationRequest(String firstName, String lastName) {}
17
18@Repository
19interface CustomerRepository extends CrudRepository<Customer, Long> {}

Dieser Controller bietet eine Funktion, um mit einem HTTP POST gegen die URL /customer einen neuen Kunden im System anzulegen. Es wird mit den Annotations @RestController und @PostMapping gearbeitet, um das Request Handling von Spring zu konfigurieren. Die Methode createNewCustomer() wird von Spring aufgerufen sobald der HTTP Body zu einem CustomerCreationRequest geparsed werden konnte. Aus diesem Request werden Vor- und Nachname des Kunden extrahiert und ein neues Customer Objekt angelegt. Dieses wird dann per save() Methode in einem Repository (und damit in einer Datenbank) persistiert und anschließend an den Aufrufer der Applikation als Ergebnis zurück gegeben.

Überlegen wir uns, was hier unter der Haube passiert. Spring bzw. der genutzte HTTP Server (z. B. Jetty) wird den eingehenden HTTP Request entgegennehmen und diesen dann einem Thread aus einem Thread-Pool zur Bearbeitung zuweisen. Dieser Thread wird zunächst versuchen, den für den Request passenden Controller zu finden und anschließend die eingehende HTTP-Nachricht von der Netzwerk-Schnittstelle lesen und parsen. Dabei handelt es sich um eine Blocking-IO-Operation, was man vor allem dann merkt, wenn der Request eine größere Datenmenge enthält (z. B. bei einem Dateiupload). Der Thread wird an diesem Punkt also in einen Wartemodus versetzt.

Sobald der gesamte Request gelesen werden konnte, wird der Thread von der JVM Runtime wieder aufgeweckt. Mit dem Request-Objekt wird dieser dann die createNewCustomer()-Methode des Controllers aufrufen und diese wiederum triggert, wie oben beschrieben, die save()-Methode des Repositorys. Innerhalb dieser Methode wird eine ausgehende Verbindung zu einer Datenbank aufgebaut, um das zu speichernde Customer-Objekt dorthin zu übertragen. Auch dabei handelt es sich um Blocking IO. Der Thread wird hier also wieder schlafen gelegt, bis das gesamte Customer-Objekt an die Datenbank übertragen wurde. Nach dem Aufwachen wird der finale HTTP Response zusammengebaut und dann wieder in einer blockierenden Operation an den Aufrufer der Applikation zurückgesendet.

Dieses einfache Beispiel zeigt einen heutzutage typischen Workload, in dem sich kurze CPU-intensive Phasen mit längeren IO-Phasen abwechseln. Bei der Abarbeitung eines Requests wird der Thread in unserem Beispiel deutlich mehr Zeit im Wartemodus verbringen, als wirklich aktiv an dem Request zu arbeiten. Und trotzdem belegt der Thread über die gesamte Zeit Systemressourcen. Wir wollen daher die Wartezeiten nutzen, um an anderen Requests zu arbeiten, und portieren unseren Controller nach Spring Webflux:

1@RestController
2class CustomerController {
3
4   private final CustomerRepository repository;
5
6   CustomerController(final CustomerRepository repository) {
7      this.repository = repository;
8   }
9
10   @PostMapping("/customer")
11   Mono<Customer> createNewCustomer(@RequestBody Mono<CustomerCreationRequest> request) {
12      return request.flatMap(dto -> 
13            repository.save(new Customer(dto.firstName(), dto.lastName()))
14      );
15   }
16}
17
18record CustomerCreationRequest(String firstName, String lastName) {}
19
20@Repository
21interface CustomerRepository extends ReactiveCrudRepository<Customer, Long> {}

Diese Implementierung ist fast identisch zur Spring-MVC-Variante von oben. Eingangsparameter und Rückgabewert der createNewCustomer()-Methode sind jetzt jedoch vom Typ Mono und auch der Body dieser Methode hat sich leicht verändert. Ansonsten erweitert das CustomerRepository jetzt ein ReactiveCrudRepository, um eine Ende-zu-Ende-Reaktivität der Anwendung zu erreichen.

Auch wenn diese Unterschiede zunächst marginal wirken, haben sie jedoch gewaltige Auswirkungen auf das mentale Modell des Entwicklers, der diesen Code schreiben soll. Dazu müssen wir uns zunächst wieder klarmachen, was genau passiert, wenn die Methode createNewCustomer() von Spring aufgerufen wird. Man lässt sich leicht dazu verleiten zu glauben, dass während der Ausführung der Methode wie oben zunächst das Request-Objekt entgegengenommen und anschließend ein neuer Customer erzeugt und persistiert wird. Das stimmt aber nicht. Schaut man genauer hin, so wird klar, dass diese Methode nichts weiter tut als Reactive Streams (genauer Monos) zusammen zu stecken. Das passiert über die Methode flatMap(). Diese Methode ruft für jedes emittierte Event des betroffenen Publishers (hier request) ein Lambda auf und ersetzt dieses Event durch all die Events, die das Lambda emittiert. Im Beispiel wird das Event CustomerCreationRequest also durch das Event ersetzt, das die save()-Methode zurückgibt. Das Ergebnis der flatMap()-Methode ist dabei wieder ein Reactive Stream, in unserem Beispiel also ein Mono<Customer>.

An dieser Stelle ist es wichtig zu verstehen, dass durch das Zusammenstecken der Monos noch keinerlei Verarbeitungslogik ausgeführt wird. Ein neuer Customer wird nicht angelegt und eine Verbindung zur Datenbank wird während der Ausführung der Methode auch nicht aufgebaut. All das passiert erst, wenn eine Subscription auf dem resultierenden Mono<Customer> erzeugt und abgefragt wird. Und genau um diese Subscription muss man sich bei der Benutzung von Spring Webflux nicht kümmern. Diese wird automatisch angelegt und Spring Webflux sorgt dafür (im Zusammenspiel mit dem Webserver, z. B. Netty), dass die gesamte Kette vom eingehenden HTTP Request bis zum ausgehenden HTTP Reponse non-blocking abgearbeitet wird. Trotz des guten Supports von Spring Webflux wird schnell klar, dass diese Form der Programmierung eine deutlich andere Art des Denkens verlangt.

Ein klassischer, imperativer Ablauf von oben nach unten wird hier durch eine Verkettung von aufeinander aufbauenden Streams ersetzt. Das bedeutet für einige Entwickler eine Umgewöhnung und damit auch erstmal eine gewisse Hürde. Zudem erzeugt dieses Programmierparadigma in vielen Fällen eine stärkere kognitive Belastung. Es sollte im Team entschieden werden, ob die Vorteile von Reactive Programming den Nachteil dieser erhöhten Komplexität rechtfertigen.

In einem nachfolgenden Artikel erläutern wir, wie an diesem Punkt Kotlin mit seinen Coroutines versucht, eine Hilfestellung zu geben, durch die Reactive Programming fast ausschließlich mit dem vertrauten imperativen Programmiermodell erreicht werden kann.

Fazit

Wir hoffen, wir konnten in diesem Beitrag die Vor- und Nachteile von Reactive Programming beleuchten und einige Hinweise darauf geben, wann man von diesem Gebrauch machen sollte und wann eher nicht. Mit dem gezeigten Beispiel sollte klar geworden sein, dass in Java ein Reactive-Ansatz wie Spring Webflux eine kognitive Belastung für die Entwickler mitbringen kann, wenn sie bisher nur ein klassisches imperatives Programmiermodell gewöhnt sind. Daher sollte gezielt abgewogen werden, in welchen Fällen Reactive Programming mit Spring Webflux zum Einsatz kommt.