MongoDB im Turbomodus mit Spring Webflux und Spring Data reactive

Um den Turbo in der Kommunikation mit der Datenbank zu zünden, gibt es Spring Data auch als reaktive Bibliothek. Damit hilft uns das reaktive Paradigma die Interaktion mit unserer Datenbank zu beschleunigen. In diesem Artikel zeige ich euch, wie man Spring Data Reactive Repositories für MongoDB benutzt.

Einführung

Spring Reactive

Zur reaktiven Programmierung mit Spring habe ich bereits 2 andere Artikel geschrieben, in denen ich hauptsächlich die Interaktion zwischen Client und Server beschreibe. Dort findet ihr auch Erklärungen warum reaktive Programmierung als skalierbare, resiliente, responsive und Event-basierte Alternative zum klassischen Ansatz meist deutlich performanter ist:

Spring's reaktiver WebClient
Spring Webflux für reaktive Webservices

Spring Data und MongoDB

Auch zu Spring Data und MongoDB habe ich bereits einen Artikel geschrieben, dort geht es aber um die klassische Programmierung:

Spring Data und MongoDB

Der Nachteil dieses Ansatzes ist, dass die Anwendung bei jeder Datenbank-Anfrage immer auf die Antwort der Datenbank warten muss. Es gibt also eine Blockade im Thread, der gerade mit der Datenbank kommuniziert. Bei reaktiven Programmieren wird diese Blockade bzw. das Warten durch eine asynchrone, Event-basierte Kommunikation vermieden. Dieses Prinzip wird auch von Spring Data Reactive und diversen noSQL Datenbanken unterstützt. Wie das genau funktioniert zeige ich euch hier.

MongoDB im Docker Container starten

Dank Docker könnt Ihr mit nur einem Befehl eine lokale Mongo Datenbank im Container starten. Diese kann dann von der Spring Anwendung verwendet werden. In diesem Blog-Artikel habe ich eine Anleitung zum Installieren von Docker hinterlegt: Spring Data und MongoDB

Wenn Docker installiert ist, könnt ihr diesen einfach Befehl zum Starten der noSQL Datenbank verwenden - den Befehl selbst habe ich auch im zuvor verlinkten Artikel erklärt:

docker run -d -p 27017:27017 --name mongodb mongo

Spring Data Reactive

Bibliotheken hinzufügen

Die Spring Data Reactive Repositories für MongoDB werden im Build-Tool, hier Gradle, mit dieser Dependency geladen:

dependencies {

implementation 'org.springframework.boot:
spring-boot-starter-data-mongodb-reactive'

implementation 'org.springframework.boot:
spring-boot-starter-webflux'

...

In meiner Demo zeige ich einen Service mit REST-API, der die Spring Data Reactive Repositories verwendet, daher benötige ich auch die Spring Webflux Bibliothek.

Verbindung zu MongoDB konfigurieren

Neben der Spring Data MongoDB Reactive Bibliothek, müssen wir in der Konfiguration noch die Verbindungs-Parameter zu unserer lokalen Datenbank hinterlegen. Das kann z.B. wie folgt in der application.properties Datei gemacht werden:

spring.data.mongodb.host=localhost

spring.data.mongodb.port=27017

spring.data.mongodb.database=local

Host, Port und Name der Datenbank ergeben sich bei mir durch die Ausführung im Docker-Container. Solltet ihr MongoDB anders bereitgestellt haben, müsst ihr eventuell noch Benutzername und Passwort in diesen Properties hinterlegen:

spring.data.mongodb.username=

spring.data.mongodb.password=

ReactiveMongoRepository und DAO Klassen

Die DAO Klasse ist, wie wir es von Spring Data kennen, eine POJO Klasse mit Annotationen:

import java.time.LocalDate;

import org.springframework.data.annotation.Id;

import org.springframework.data.mongodb.core.mapping.Document;

@Document(collection = "Employee")

public class EmployeeDAO {

@Id private String id;

private String empNo;

private String fullName;

private LocalDate hireDate;

// Getters and Setters for all attributes

}

Es sind dieselben Spring Data Annotations-Klassen, die ich auch schon beim klassischen, nicht reaktiven Spring Data Tutorial verwendet habe. @Document und @Id habe ich daher schon hier erklärt: Spring Data und MongoDB.

Das Repository, welches die Schnittstelle zur Datenbank implementiert, definieren wir wie schon beim ersten Tutorial nur als Interface:

import org.springframework.data.mongodb.repository.ReactiveMongoRepository;

import reactor.core.publisher.Flux;

//String: Type of Employee ID.

public interface ReactiveEmployeeRepository

extends ReactiveMongoRepository<EmployeeDAO, String> {

Flux<EmployeeDAO> findAllByHireDateGreaterThan(

LocalDate hireDate);

}

ReactiveMongoRepository ist das Interface für reaktive Spring Data MongoDB Repositories, welches wir erweitern müssen. Durch dieses Interface erben wir diverse Lese- (find) und Schreib-Methoden (insert), die mit den reaktiven Datenstrom-Klassen Mono und Flux arbeiten. Spring Data Reactive Repositories stellt uns also automatisch reaktive Implementierungen dieser Methoden bereit.
Wie wir es von Spring Data kennen, können wir auch hier wieder eigene Methoden mit einer Kette von Schlüsselwörtern als Namen definieren. Der Methodename findAllByHireDateGreaterThan besteht aus solchen Schlüsselwörtern und definiert Flux als Rückgabewert. Spring Data kann nun mit seinem Query Derivation Mechanismus anhand der Schlüsselwörter, des Rückgabe-Typs und der Parameter eine reaktive Implementierung dieser Methode automatisch bereitstellen. Konkret liefert diese Methode einen Datenstrom mit allen EmployeeDAO Objekten, die nach dem übergebenen Datum (LocalDate) angestellt wurden.

Reaktiver Controller

Wie ein reaktiver RestController genau funktioniert, habe ich bereits in diesem Blog Artikel gezeigt: Spring Webflux für reaktive Webservices. Das Besondere am hier gezeigt RestController ist, dass er eine andere reaktive Komponente, unser ReactiveEmployeeRepository, aufruft und dessen reaktive Antwort direkt als Flux oder Mono an den Client weitergibt. Der RestController und unsere MongoDB Repository sind also durchgängig reaktiv! Ansonsten hat der RestContoller keine Besonderheiten, die ich nicht schon in anderen verlinkten Blog-Artikeln vorgestellt habe:

@RestController
@RequestMapping(Constants.URL_PATH_EMPLOYEE)
public class ReactiveEmployeePersistenceController {
   @Autowired private ReactiveEmployeeRepository repo;

   @GetMapping
   @ResponseStatus(HttpStatus.OK)
   public Flux<EmployeeDAO> readDbEntries(
           @RequestParam @DateTimeFormat(iso = ISO.DATE)
           LocalDate hiredAtOrLater) {
return repo.findAllByHireDateGreaterThan(hiredAtOrLater);
   }

   @DeleteMapping
   @ResponseStatus(HttpStatus.OK)
   public Mono<Void> deleteAllEmployees() {
return repo.deleteAll().doOnSuccess(ignored -> log.info(

"All database entries deleted."));
}

...

}

Um den Demo-Code möglichst kompakt zu halten, rufe ich aus dem Controller direkt das Repository auf - in echten Projekten solltet Ihr das nicht tun, da eine Trennung entsprechend Schichten eine bessere Architektur ist.

Test der reaktiven Datenbank-Verbindung

Mein Test soll in erster Linie zeigen, dass durch die reaktive Datenbank-Interaktion keine Threads blockiert werden. In der Praxis finden wir häufig die 3 Komponenten Client, Server und Datenbank. Der folgende JUnit Test agiert wie ein Client, indem Anfragen per Spring WebClient mittels HTTP Request an den Server geschickt werden. Der Server ist hier in der Demo sehr schlank aufgebaut und hat nur einen reaktiven Rest-Controller und das reaktive Repository, um die NoSQL Mongo Datenbank aufzurufen. Die 3 Komponenten mit den jeweiligen Klassen werden im folgenden Diagramm visualisiert. Alle Komponenten und Klassen werden vom folgenden JUnit Test benutzt.

Im ReactiveEmployeePersistenceControllerTest benutzte Komponenten und Klassen

@SpringBootTest(webEnvironment = WebEnvironment.DEFINED_PORT)

class ReactiveEmployeePersistenceControllerTest {

@Autowired private ReactiveEmployeeRepository repo;

@BeforeEach

void setup() {

repo.deleteAll().block();

repo.insert(new EmployeeDAO(null, "007",

"James Bond", LocalDate.EPOCH.plusDays(1))).block();

repo.insert(new EmployeeDAO(null, "001",

"Jane Moneypenny", LocalDate.EPOCH)).block();

}

@Test

void reactiveDeleteTest() {

Mono<Void> mono = WebClient.create("http://localhost:8080")
.delete().uri("/employee")

.retrieve().bodyToMono(Void.class);

assertEquals(2, repo.count().block());

final Disposable disposable = mono.subscribe();

System.out.println("Delete request was sent to database");

Awaitility.await().until(() -> disposable.isDisposed());

assertEquals(0, repo.count().block());

}

...

}

Der Ausschnitt zeigt nur einen Testfall, der den komplette Inhalt der Collection "Employee" in der MongoDB löscht. Weitere ähnlich aufgebaute Testfälle findet Ihr in meinem GitHub Projekt zusammen mit dem restlichen Code: https://github.com/elmar-brauch/webflux

@SpringBootTest startet die komplette reaktive Spring Boot Anwendung und veröffentlicht die API über den default Port 8080.
Das reaktive Repository hole ich per Dependency Injection in den Test (@Autowired), damit ich nach jedem API Aufruf den Zustand in der MongoDB mittels von Spring Data generiertem Code prüfen kann.
Vor jedem Test (siehe mit @BeforeEach annotierte setup Methode) lösche ich den kompletten Inhalt der Datenbank und mache dann 2 Einträge, um Seiteneffekte aus anderen Tests zu verhindern. Da es mir hier um das zeigen der reaktiven Datenbank-Interaktion geht, habe ich es so gemacht. Würde ich eine richtige Anwendung entwickeln, würde ich die Tests anders strukturieren und mit Mocks oder einer eingebetteten MongoDB (siehe dazu https://www.baeldung.com/spring-boot-embedded-mongodb) arbeiten.
In der setup Methode habe ich jede Datenbank Schreiboperation blockierend mit block abgeschickt. Würde ich das reaktiv mit subscribe statt block machen, wäre unklar in welchem Zustand die Datenbank zu Beginn der eigentlichen Tests wäre, da die 3 Anfragen durch die Asynchronität in beliebiger Reihenfolge ausgeführt werden können.
Das Versenden von Requests mit dem WebClient hatte ich hier schon erklärt: Spring's reaktiver WebClient. In meinem Blog ist die Rückgabe der subscribe Methode vom Typ Disposable neu. Das Disposable Interface bietet 2 Methoden an: dispose zum Beseitigen des Tasks bzw. des HTTP Requests und isDisposed zum Prüfen, ob der Task fertig ist.
Nachdem der HTTP Request abgeschickt wurde, warte ich mit Awaitility so lange bis das zugehörige Disposable im Zustand "disposed" ist bzw. bis isDisposed true liefert. Dazu verwende ich die Bibliothek Awaitility, welche ich in diesem Artikel vorgestellt habe: Awaitility. Danach prüfe ich nur noch, ob die Einträge in der Datenbank-Collection wirklich gelöscht wurden.
Sowohl im Controller ReactiveEmployeePersistenceController als auch im JUnit Test schreibe ich einen Log-Eintrag. Die Reihenfolge kann durch die Asynchronität beliebig sein, in den meisten Fällen sollte aber zuerst "Delete request was sent to database" geloggt werden und danach die Log-Meldung des Controllers erscheinen. Diese Reihenfolge zeigt, dass der Main-Thread in dem der JUnit-Test ausgeführt wird, nicht blockiert ist und wir somit eine reaktive Lösung haben.

Fazit

In diesem Artikel habe ich gezeigt, wie eine reaktive Kommunikationsstrecke vom Client über den Server bis zur noSQL Datenbank aussieht. Dank durchgehender, nicht blockierender, reaktiver Programmierung können wir die CPUs bei entsprechend großen Mengen von Benutzeranfragen voll auslasten und somit eine sehr hohe Performance erreichen.

Reaktive Programmierung ist sicherlich etwas komplexer als die klassische. Das Spring Framework bzw. Spring reactive hilft uns diese Komplexität zu managen. Falls ihr noch nicht so fit bezüglich der in Java 8 eingeführten Konzepte: funktionale Interfaces, Lambda und Stream API seid, empfehle ich euch diese Themen zu vertiefen, dann sollte euch die reaktive Programmierung leichter fallen.

Ich freue mich über Fragen, Anmerkungen und Kommentare.

Wie immer findet ihr den Code zu diesem Artikel in GIT:

https://github.com/elmar-brauch/webflux

Dieses Blog durchsuchen

Agile Coding mit Java, Kotlin, Spring und Microservices