Reaktive Programmierung mit Spring Webflux

In diesem Blog-Beitrag möchten wir einen kurzen Überblick über Spring Webflux geben. Zusätzlich gibt der Beitrag eine kleine Starthilfe mit einer Auswahl an Codebeispielen, damit ihr erfolgreich in das Projekt einsteigen könnt.

In Wikipedia wird reaktive Programmierung wie folgt beschrieben:

Das zugrundeliegende Ausführungsmodell propagiert Änderungen in den Datenflüssen automatisch. Seit Version 5 unterstützt das Spring Framework reaktive Programmierung und bietet reaktive Implementierungen für Webanwendungen, Datenbankzugriffe, Sicherheit und streambasierte Datenverarbeitung.

Klassische Programmiermodelle wie beispielsweise Spring MVC verwenden für jede Anfrage einen Thread und belegen diesen so lange, bis die aktuelle Aufgabe abgeschlossen ist und geben ihn dann wieder frei. Muss während der Codebearbeitung z.B. eine Datenbank oder ein entferntes System angesprochen werden, bei dem eine langsame Antwortzeit zu erwarten ist, muss der Thread lange blockiert werden. Um die Antwortbereitschaft aufrecht zu erhalten, wird oft ein größerer Threadpool vorgehalten. Je nach Szenario kann ein Java-Thread auch schnell 1MB Speicher allokieren, was insbesondere in Cloud-Umgebungen schnell zu erhöhten Kosten führt.

Project Reactor

Das Spring Framework verwendet das Open-Source-Projekt Reactor als Basis für die Implementierung reaktiver Programmierung. Reactor ist eine nicht-blockierende reaktive Open-Source-Programmierumgebung für die Java Virtual Machine, die auf der Reactive-Streams-Spezifikation basiert. Es setzt direkt auf den funktionalen APIs von Java 8 auf und verwendet durchgängig CompletableFuture, Stream und Duration. Zusätzlich unterstützt Reactor eine nicht-blockierende Interprozesskommunikation mit dem reactor-netty-Projekt, das eine backpressure-fähige Netzwerk-Engine für HTTP bereitstellt. Die Reactive Streams Spezifikation sieht eine gewisse Standardisierung für die JVM aber auch für Javascript vor und basiert auf folgenden Schnittstellen:

• Subscriber: Der Subscriber abonniert einen Publisher und wird durch Callbacks über Neuerungen informiert.
• Subscription: Die Subscription beschreibt das Verhältnis zwischen Subscriber und Publisher.
• Publisher: Der Publisher ist für die Veröffentlichung der Daten an die abonnierten Subscriber verantwortlich.
• Processor: Ein Processor transformiert Elemente, die zwischen Publisher und Subscriber übertragen werden.

Das Reactor-Projekt bietet zwei Implementierungen des Publisher-Interfaces, Mono und Flux, die in den folgenden Beispielen häufig verwendet werden. Flux ist als asynchrone Sequenz von 0-N Elementen implementiert und Mono als 0-1 Element.

Wie arbeitet man mit Spring Webflux?

Die Reactor API bietet eine sehr große Anzahl von Methoden. Es gibt zwar eine Art Tutorial, aber das kann gerade beim Einstieg in Webflux überwältigend sein.

Es gibt bereits viele Tutorials auf den üblichen Webseiten, wie man mit Spring Webflux arbeiten kann. Diese beschränken sich jedoch oft auf die Kommunikation nach außen, also auf Controller, Datenbank-Repositories und Webclients, die andere REST-APIs konsumieren. Unserer Erfahrung nach ist dies aber eher der unproblematischere Teil, da hier ein Großteil der Arbeit bereits damit getan ist, die jeweils relevanten Dependencies durch Spring Webflux zu ersetzen und die Methodensignaturen entsprechend anzupassen. Die Konvertierung einer Controller-Methode könnte etwa so aussehen:

	
	//Spring Web MVC
	@GetMapping("/{id}")
	private Blogpost getBlogpostById(@PathVariable String id) {
	    return blogpostService.findBlogpostById(id);
	}
	//Spring Webflux
	@GetMapping("/{id}")
	private Mono<Blogpost> getBlogpostById(@PathVariable String id) {
	    return blogpostService.findBlogpostById(id);
	}
	

In diesem Fall registriert sich der Rest-Controller automatisch als Subscriber auf die entsprechende Service-Methode und schreibt das Ergebnis wie gewohnt in den Body der Response des HTTP-Requests. Voraussetzung ist natürlich, dass der Blog-Post-Service auch ein Mono statt eines einfachen Blog-Posts zurückgibt. Und hier wird es unserer Meinung nach spannend. Wie gehe ich mit Mono und Flux auf Service-Ebene um und wie implementiere ich Geschäftslogik, die über das einfache Weiterreichen von Daten hinausgeht?

Das ist der Schwerpunkt dieses Abschnitts. In den Beispielen wird nur mit Monos gearbeitet, aber fast alle genannten Methoden sind auch für Flux verfügbar. Eine Ausnahme bildet die Methode zipWhen, die nur für Monos implementiert ist.

map und flatMap

Diese Methoden sind jedem bekannt, der schon einmal mit Java Streams oder funktionalen Programmiersprachen gearbeitet hat. Sie sollten immer dann verwendet werden, wenn man das Ergebnis eines Monos oder Fluxes weiterverarbeiten möchte. Angenommen, ich habe in einem Request die Daten eines Users in einer API abgefragt, benötige im Folgenden aber nur das Kürzel, das aus Vor- und Nachname gebildet werden kann.

	
	public Mono<String> getAbbreviatedName(String userId) {
	    Mono<UserData> userDataMono = userClient.getUser(userId);
	    Mono<String> abbreviatedNameMono = userDataMono.map(userData -> buildAbbreviatedName(userData));
	    return abbreviatedNameMono;
	}
	private String buildAbbreviatedName(UserData userData){[...]}
	

In der innerhalb des map-Befehls aufgerufenen Methode (hier buildAbbreviatedName) muss nicht mit Webflux Publishern gearbeitet werden, die eigentliche Geschäftslogik kann also mit gewöhnlichem Java-Code implementiert werden.

Der Operator flatMap unterscheidet sich von map dadurch, dass er als Rückgabewert der übergebenen Methode wieder einen Publisher erwartet. Dies ist dann sinnvoll, wenn diese Methode ihrerseits wieder einen API-Aufruf absetzen soll. In unserem Beispiel könnte ein Blog-Eintrag mit den Daten des Users erstellt werden.

	
	public Mono<BlogpostWithAuthor> saveBlogpost(Blogpost blogpost, String userId) {
	    Mono<UserData> userDataMono = userClient.getUser(userId);
	    Mono<BlogpostWithAuthor> savedBlogpostMono = userDataMono.flatMap(userData -> saveBlogpost(blogpost, userData));
	    return savedBlogpostMono;
	}
	private Mono<BlogpostWithAuthor> saveBlogpost(Blogpost blogpost, UserData userData){[...]}
	

zipWhen und zipWith

Wir können also mit map und flatMap die Ergebnisse eines Publishers transformieren. Was aber, wenn die Daten von zwei oder mehr Publishern kombiniert benötigt werden? In diesem Fall können die Methoden zipWith und zipWhen verwendet werden. Im folgenden Beispiel wird die Methode zipWith verwendet, um die Daten eines Blog Posts und eines Users zu kombinieren und daraus ein BlogpostWithAuthor zu erstellen.

	
	public Mono<BlogpostWithAuthor> getBlogpost(String blogpostId, String authorId) {
	    Mono<Blogpost> blogpostMono = blogpostClient.getBlogpost(blogpostId);
	    Mono<UserData> authorMono = userClient.getUser(authorId);
	    Mono<Tuple2<Blogpost, UserData>> blogpostAuthorMono = blogpostMono.zipWith(authorMono);
	    Mono<BlogpostWithAuthor> result = blogpostAuthorMono.map(blogpostUserTuple -> 
	            buildBlogpostWithAuthor(blogpostUserTuple.getT1(), blogpostUserTuple.getT2()));
	    return result;
	}
	private BlogpostWithAuthor buildBlogpostWithAuthor(Blogpost blogpost, UserData userData){[...]}
	

In diesem Szenario ist es auch denkbar, dass die authorId nicht explizit übergeben werden muss, sondern Teil des Blogpost-Objekts ist. In diesem Fall müsste der UserClient mit Informationen aus der Response des BlogpostClients aufgerufen werden, um dann mit beiden Ergebnissen weiterzuarbeiten. Hier kommt zipWhen ins Spiel. Mit dieser Methode kann man, ähnlich wie mit flatMap, das Ergebnis eines Publishers transformieren, mit dem Unterschied, dass man anschließend wieder ein Tuple2 mit dem ursprünglichen und dem transformierten Ergebnis erhält.

	
	public Mono<BlogpostWithAuthor> getBlogpost(String blogpostId) {
	    Mono<Blogpost> blogpostMono = blogpostClient.getBlogpost(blogpostId);
	    Mono<Tuple2<Blogpost, UserData>> blogpostAuthorMono = blogpostMono.zipWhen(blogpost -> 
	            userClient.getUser(blogpost.getAuthorId()));
	    Mono<BlogpostWithAuthor> savedBlogpostMono = blogpostAuthorMono.map(blogpostUserTuple -> 
	            buildBlogpostWithAuthor(blogpostUserTuple.getT1(), blogpostUserTuple.getT2));
	    return savedBlogpostMono;
	}
	private Mono<BlogpostWithAuthor> buildBlogpostWithAuthor(Blogpost blogpost, UserData userData){[...]}
	

doOnNext und delayUntil

Die bisher vorgestellten Methoden sind alle primär darauf ausgerichtet, die Werte eines Editors in irgendeiner Weise zu transformieren. Es gibt aber auch Methoden, mit denen Seiteneffekte ausgeführt werden können, die den Wert des Publishers nicht beeinflussen. Zum Beispiel könnten wir in einer getBlogpost-Methode zusätzlich einen Zähler erhöhen, der anzeigt, wie oft der Post abgerufen wurde.

	
	public Mono<Blogpost> getBlogpost(String blogpostId) {
	    Mono<Blogpost> blogpostMono = blogpostClient.getBlogpost(blogpostId);
	    return blogpostMono.doOnNext(blogpost -> incrementViewCount(blogpost));
	}
	private Mono<Integer> incrementViewCount(Blogpost blogpost){[...]}
	

Hierbei ist zu beachten, dass sich zwar am eigentlichen Inhalt des Monos nichts ändert, der Rückgabewert der Methode doOnNext aber weiterhin verwendet werden muss, damit die Methode auch Teil der Ausführungssequenz wird.

Außerdem ist der Aufruf von incrementViewCount wieder asynchron. Das bedeutet, dass nicht davon ausgegangen werden kann, dass der Zähler aktuell ist, wenn getBlogpost sein Ergebnis zurückgibt. Wenn dies erforderlich ist, muss delayUntil anstelle von doOnNext verwendet werden.

onErrorMap, onErrorReturn und doOnError

Schließlich ist noch die Fehlerbehandlung zu erwähnen. Wenn in der Ausführungssequenz eine Methode einen Fehler wirft, gibt der Publisher anstelle des Wertes ein Error-Signal aus. Dieses Signal wird wie eine normale RuntimeException an den nächsten Subscriber weitergereicht, bis es irgendwo explizit behandelt wird.

Mit onErrorMap kann eine Exception abgefangen und eine andere Exception geworfen werden. Mit onErrorReturn kann beim Auftreten einer Exception ein Standardwert zurückgegeben werden, onErrorResume ruft einen alternativen Publisher auf. Jeder dieser Methoden kann auch die Klasse der zu behandelnden Exception übergeben werden, um die zu behandelnden Exceptions einzuschränken.

Wenn etwa in der Methode Shortcut aus dem ersten Beispiel dieses Kapitels einfach der Standardstring “Unknown” zurückgegeben werden soll, wenn kein Benutzer mit der angegebenen ID existiert, könnte die Implementierung wie folgt aussehen:

	
	public Mono<String> getAbbreviatedName(String userId) {
	    Mono<UserData> userDataMono = userClient.getUser(userId);
	    Mono<String> abbreviatedNameMono = userDataMono.map(userData -> buildAbbreviatedName(userData));
	    return abbreviatedNameMono.onErrorReturn(UserNotFoundException.class, "Unbekannt");
	}
	private String buildAbbreviatedName(UserData userData){[...]}
	

Fazit

Mit diesen 9 Methoden solltet ihr für den Einstieg in die Programmierung mit Spring Webflux gerüstet sein. Solltet ihr Anwendungsfälle haben, die hier nicht abgedeckt wurden, können wir euch die Operatorenübersicht aus dem Project-Reactor-Handbuch empfehlen, die bereits eingangs im letzten Kapitel erwähnt wurde.

Mit etwas Einarbeitungszeit wird die reaktive Programmierung immer einfacher und intuitiver, aber es lässt sich auch nicht leugnen, dass sie zu teilweise komplizierterem und schlechter lesbarem Code führt. Daher sollte man sich vor Beginn eines Projekts gut überlegen, ob die Vorteile, die Spring Webflux mit sich bringt, diese zusätzliche Komplexität aufwiegen.