Reactive Spring

Reactive Spring — WebFlux & RSocket

Modul 11 hat gezeigt, wie Virtual Threads blocking Code skalieren. Modul 12 dreht den Spieß um: kein Thread pro Request, sondern wenige Event-Loop-Threads, die mit nicht-blockierendem I/O tausende Verbindungen gleichzeitig bedienen. Werkzeuge: Project Reactor (Flux/Mono), WebFlux, RSocket und R2DBC — plus die ehrliche Frage, wann sich der Stilwechsel lohnt und wann Virtual Threads die bessere Wahl bleiben.

Lektion 56: Reactive Paradigma & Project Reactor (Flux/Mono)

Reactive heißt: Der Datenfluss wird deklarativ als Pipeline beschrieben — nicht "rufe auf und warte", sondern "wenn Daten ankommen, tu dies damit". Grundlage ist der Reactive-Streams-Vertrag: Ein Publisher erzeugt Items, ein Subscriber meldet sich an und gibt über Demand an, wie viel er verarbeiten kann — das ist Backpressure. Project Reactor ist die Spring-Implementierung dieses Vertrags mit zwei Typen:

// Mono<T>: 0..1 Element — Ergebnis oder Fehler
Mono<Policy> policy = policyRepository.findById(insuredId);

// Flux<T>: 0..N Elemente — Stream, Liste, SSE-Feed
Flux<Claim> claims = claimRepository.findAllByInsured(insuredId);

// Nichts läuft bis hierhin — nur die Pipeline ist "zusammengebaut" (Assembly)
claims.filter(Claim::isOpen)
    .map(ClaimMapper::toDto)
    .subscribe(dto -> notifications.send(dto)); // erst jetzt fließen Daten

Der entscheidende Unterschied zu einer Liste oder Stream: Die Pipeline ist lazy. findById() liefert kein Policy-Objekt, sondern eine Beschreibung, wie man eines bekommen würde. Erst subscribe() startet den Datenfluss, und der Subscriber kann mit request(n) steuern, wie viele Items er sofort abnehmen will — der Publisher darf dadurch nicht schneller liefern, als der Konsument verarbeitet.

Und die ehrliche Ansage: Reactive ist nicht automatisch schneller. Ein einzelner Request wird nicht kürzer — dieselbe DB-Abfrage dauert gleich lange. Der Gewinn liegt im Durchsatz bei vielen gleichzeitigen Verbindungen: Statt einem wartenden Thread (1 MB Stack, Context-Switch-Kosten) pro Request bedienen wenige Event-Loop-Threads alle Verbindungen und nutzen die I/O-Wartezeit für andere Requests. CPU-gebundene Arbeit wird durch Reactive nicht schneller — nie. Der Unterschied zu Virtual Threads (Modul 11): VTs machen blockierenden Code billig, Reactive macht Warten strukturiert nicht-blockierend — mit deutlich mehr kognitiver Last im Code.

Lektion 57: WebFlux — annotierte Controller & funktionales Routing

spring-boot-starter-webflux statt spring-boot-starter-web tauscht die Laufzeit aus: Netty statt Tomcat, Event-Loop-Threads (etwa so viele wie CPU-Kerne) statt Thread-per-Request. Die annotated-Controller-Welt kennt man weiterhin — nur die Rückgabetypen sind jetzt Mono/Flux:

@RestController
public class ClaimController {

    @GetMapping("/claims/{id}")
    public Mono<ClaimDto> claim(@PathVariable String id) {
        return claimService.find(id).map(ClaimMapper::toDto);
    }

    @GetMapping(value = "/claims/stream",
        produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public Flux<ClaimDto> openClaims() {
        return claimService.openClaims(); // SSE: jedes neue Element = Event
    }
}

Wer es expliziter mag, definiert Routen funktional — RouterFunctions trennen Routing (wohin?) vom Handling (was passiert?):

@Bean
RouterFunction<ServerResponse> routes(ClaimHandler handler) {
    return RouterFunctions.route()
        .GET("/claims/{id}", handler::getById)
        .GET("/claims", handler::findAll)
        .build();
}

Das Threading-Modell ist der Kernunterschied zu Spring MVC: Kommt ein Request an, übernimmt ein Event-Loop-Thread die Bearbeitung bis zum ersten nicht-blockierenden Punkt und springt dann zum nächsten Request. Läuft auf diesem Event-Loop-Thread dennoch blocking Code (JDBC, block(), Thread.sleep), stehen alle anderen Requests — der Event-Loop ist eine gemeinsam genutzte Ressource. Getestet wird idiomatisch mit WebTestClient, dem reaktiven Geschwister zu TestRestTemplate/MockMvc:

// Slice-Test für einen WebFlux-Controller
WebTestClient client = WebTestClient.bindToController(new ClaimController()).build();

client.get().uri("/claims/C-1").exchange()
    .expectStatus().isOk()
    .expectBody(ClaimDto.class)
    .consumeWith(r -> assertThat(r.getResponseBody().id()).isEqualTo("C-1"));

Lektion 58: RSocket — bidirektionale Streams & Backpressure

HTTP kennt nur ein Muster: Request kommt, Response geht — Client fragt, Server antwortet. RSocket ist ein binäres Protokoll (über TCP oder WebSocket) mit vier Interaktionsmodellen, und Spring bildet jedes direkt auf Reactor-Typen ab:

Serverseitig genügt eine @MessageMapping-Methode — das Pendant zu @GetMapping, nur über das RSocket-Routing:

@Controller
public class DwdRsocketController {

    @MessageMapping("dwd.trockenindex")
    public Flux<DwdIndex> stream(String region) {
        return dwdService.liveIndex(region); // Flux mit Backpressure
    }
}

# application.yml — RSocket-Server über WebSocket exponieren
spring:
  rsocket:
    server:
      transport: websocket
      mapping-path: /rsocket

Zwei Protokoll-Features heben RSocket über rohe WebSockets: Leasing lässt den Server die Anfragenrate pro Client begrenzen (Flow Control, nicht der Client limitiert sich selbst), und Resumption stellt eine unterbrochene Session wieder her — Streams laufen nach dem Reconnect weiter, ohne Daten zu verlieren. Wann RSocket statt HTTP/REST? Wenn es wirklich bidirektional wird: Service-zu-Service-Streaming mit Backpressure, Clients auf mobilen oder instabilen Verbindungen, Live-Feeds in beide Richtungen. Für klassisches CRUD-HTTP bleibt REST die einfachere Wahl — Browser-Support für RSocket ist umständlich, und REST kann man mit curl debuggen.

Lektion 59: Reactive Data mit R2DBC & Transaktionen

Eine WebFlux-Anwendung mit JDBC ist halb reactive: Der Controller gibt Mono/Flux zurück, aber die DB-Abfrage blockiert den Event-Loop-Thread. R2DBC (Reactive Relational Database Connectivity) ist die konsequente Antwort — eine Reactive-Streams-API für relationale Datenbanken mit nicht-blockierenden Treibern (PostgreSQL, H2, MySQL, MariaDB). Statt JPA- Repositories arbeitet man mit DatabaseClient:

@Service
public class DwdService {

    private final DatabaseClient database;

    DwdService(DatabaseClient database) {
        this.database = database;
    }

    public Flux<DwdIndex> criticalRegions(double minIndex) {
        return database.sql("""
                SELECT region, value FROM dwd_index WHERE value > :min
                """)
            .bind("min", minIndex)
            .map((row, meta) -> new DwdIndex(
                row.get("region", String.class),
                row.get("value", Double.class)))
            .all(); // Flux<DwdIndex>, lazily aus der DB gestreamt
    }
}

Transaktionen funktionieren weiterhin deklarativ — aber der ReactiveTransactionManager führt die Transaktion durch die Pipeline: Beginn beim Subscription-Start, Commit nach onComplete, Rollback bei Fehler. Dazu muss die Methode selbst Mono/Flux zurückgeben; eine blocking Methode mit @Transactional bekommt keinen reaktiven Transaction-Manager:

@Service
public class ClaimService {

    @Transactional
    public Mono<Claim> fileClaim(ClaimDto dto) {
        return claimRepository.save(Claim.from(dto))
            .flatMap(claim -> auditLog.record(claim).thenReturn(claim));
    } // Commit erst, wenn der Mono-Subscriber das onComplete bekommt
}

Die Migration von JPA zu R2DBC ist eine bewusste Entscheidung: Hibernate ist durch und durch sync — Lazy Loading, Dirty Tracking, Persistence Context brauchen blocking I/O, und für R2DBC gibt es kein JPA. Pragmatische Strategie: Repositories schichtweise migrieren, dabei die mächtigen JPA-Features (Assoziationen, Locking) gegen eigene, flache SQL-Queries tauschen. Einfache Queries überleben das trivial; wer Relationsgraphen lazily lädt, muss das Datenmodell denormalisieren oder aggressive Joins schreiben.

Lektion 60: Migration — Blocking zu Reactive, und wann NICHT

Blocking Code schleicht sich gern zurück in reaktive Pipelines — ein block() hier, ein JDBC-Call dort. BlockHound macht das sichtbar: Es instrumentiert die JVM und wirft eine Exception, sobald blocking Code auf einem Event-Loop-Thread landet — ideal als Test-Wächter:

// build.gradle: testImplementation "io.projectreactor.tools:blockhound"
@TestConfiguration
static class BlockhoundConfig {
    @BeforeAll
    static void install() {
        BlockHound.install(); // block() auf dem Event-Loop → Test schlägt fehl
    }
}

Die klassischen Anti-Patterns und ihre Fixes:

// ANTI-PATTERN 1: block() im reaktiven Fluss
Policy p = policyRepository.findById(id).block(); // Event-Loop lahmgelegt

// FIX: Pipeline durchreichen; blocking Code auf boundedElastic kanalisieren
Mono<Policy> policy = Mono.fromCallable(() -> legacyClient.load(id))
    .subscribeOn(Schedulers.boundedElastic()); // eigener Pool, Event-Loop bleibt frei

// ANTI-PATTERN 2: ThreadLocal für Request-Kontext (Security, Tracing)
// Auf Event-Loops wechselt der Thread ständig — ThreadLocal ist unzuverlässig.
// FIX: Reactor Context — der Kontext fließt mit den Signalen, nicht mit dem Thread
Mono<UUID> userId = Mono.deferContextual(ctx ->
    Mono.just(ctx.get("userId")));

Und die ehrliche Entscheidungsmatrix zum Abschluss des Kurses:

Reactive lohnt sich, wenn I/O-Wartezeit der Engpass ist und viele Verbindungen gleichzeitig bedient werden müssen — oder wenn Streams mit Backpressure das Fachliche sind. Für den klassischen request/response-Stack mit JDBC bleiben Virtual Threads der einfachere Weg: gleiche Skalierung, lesbarer Code, normale Stacktraces. Beide Modelle koexistieren — blocking Teile via boundedElastic einkapseln, den Rest reaktiv halten.