Reactive Programming in Java: Reactive Streams, Reactor e WebFlux

“`html

La Reactive Programming è un paradigma di programmazione pensato per gestire flussi di dati asincroni, eventi e operazioni che cambiano nel tempo. Invece di chiedere continuamente se un dato è disponibile, l’applicazione definisce una catena di elaborazione e reagisce quando il dato viene prodotto.

Questo approccio è particolarmente utile nelle applicazioni che devono gestire numerose operazioni di input/output, molte connessioni contemporanee, eventi continui o servizi remoti con tempi di risposta variabili.

Nel mondo Java, la programmazione reattiva è collegata soprattutto alla specifica Reactive Streams, alla classe java.util.concurrent.Flow, a librerie come Project Reactor e RxJava e a framework web non bloccanti come Spring WebFlux.

🔄 Che cos’è la Reactive Programming

La programmazione reattiva è un paradigma basato sui flussi di dati e sulla propagazione dei cambiamenti. I dati possono arrivare immediatamente, dopo alcuni secondi oppure continuare a essere prodotti senza una fine prestabilita.

Un flusso reattivo può rappresentare, per esempio:

• il risultato di una richiesta HTTP;
• una sequenza di record letti da un database;
• i messaggi ricevuti da una coda;
• gli eventi generati da un’interfaccia utente;
• le letture continue di un sensore;
• gli aggiornamenti di prezzo di un prodotto;
• le notifiche inviate da un sistema;
• un flusso di dati proveniente da una WebSocket.

In un programma imperativo tradizionale, il codice richiede un valore e attende il risultato:

Product product = productService.findById(10L);

System.out.println(product.name());

In un modello reattivo, il servizio restituisce un oggetto che rappresenta un risultato futuro:

Mono<Product> product =
        productService.findById(10L);

product.subscribe(
        value -> System.out.println(value.name())
);

Il metodo non deve necessariamente attendere che il prodotto sia disponibile. Restituisce immediatamente una descrizione della futura elaborazione.

⚖️ Programmazione sincrona, asincrona e reattiva

I tre termini vengono spesso utilizzati come sinonimi, ma descrivono concetti differenti.

Un’operazione può essere asincrona ma non reattiva. Un CompletableFuture, per esempio, rappresenta normalmente un singolo risultato futuro, mentre un publisher reattivo può rappresentare zero, uno, molti o infiniti elementi.

Codice sincrono

public Product loadProduct(long id) {

    /*
     * Il thread rimane in attesa fino al completamento
     * della chiamata al database.
     */
    return repository.findById(id);
}

Codice asincrono con CompletableFuture

public CompletableFuture<Product> loadProduct(
        long id
) {
    return CompletableFuture.supplyAsync(
            () -> repository.findById(id)
    );
}

Codice reattivo

public Mono<Product> loadProduct(long id) {

    /*
     * Il repository deve essere realmente reattivo
     * affinché l'intera catena resti non bloccante.
     */
    return repository.findById(id);
}

🏗️ Reactive Programming e Reactive Systems

La programmazione reattiva e i sistemi reattivi sono collegati, ma non sono la stessa cosa.

La programmazione reattiva riguarda principalmente il modo in cui il codice gestisce flussi, eventi e propagazione dei cambiamenti.

Un sistema reattivo, secondo i principi del Reactive Manifesto, mira invece a essere:

• responsive: risponde in tempi prevedibili;
• resilient: continua a rispondere anche in presenza di errori;
• elastic: si adatta alle variazioni del carico;
• message-driven: utilizza messaggi asincroni tra componenti.

Utilizzare Flux e Mono non rende automaticamente reattiva l’intera architettura. Un’applicazione può utilizzare operatori reattivi ma continuare ad avere:

• database bloccanti;
• dipendenze sincrone;
• un unico punto di errore;
• assenza di timeout;
• nessuna strategia di recupero;
• scarsa capacità di adattarsi al carico.

🌊 Che cosa sono i Reactive Streams

Reactive Streams è una specifica progettata per standardizzare l’elaborazione asincrona di sequenze di dati con backpressure non bloccante.

La specifica non definisce un linguaggio completo di operatori come map, filter o flatMap. Stabilisce invece un insieme minimo di interfacce e regole attraverso le quali produttori e consumatori possono comunicare.

La sua struttura può essere rappresentata così:

Producer
   │
   ▼
Publisher
   │
   │ onSubscribe
   ▼
Subscriber
   │
   ├── request(n)
   └── cancel()

Flusso dei dati:       Publisher  ─────────► Subscriber
Segnale di domanda:    Publisher  ◄───────── Subscriber

Il produttore pubblica gli elementi, mentre il consumatore comunica quanti elementi è pronto a ricevere.

🧩 Publisher, Subscriber, Subscription e Processor

La specifica Reactive Streams definisce quattro componenti fondamentali.

Publisher

Il Publisher rappresenta il produttore degli elementi. Permette a un subscriber di iscriversi al flusso.

public interface Publisher<T> {

    void subscribe(
            Subscriber<? super T> subscriber
    );
}

Subscriber

Il Subscriber riceve i segnali inviati dal publisher.

public interface Subscriber<T> {

    void onSubscribe(Subscription subscription);

    void onNext(T item);

    void onError(Throwable throwable);

    void onComplete();
}

Subscription

La Subscription rappresenta il collegamento tra publisher e subscriber. Permette al consumatore di richiedere elementi oppure di annullare la sottoscrizione.

public interface Subscription {

    void request(long numberOfItems);

    void cancel();
}

Processor

Un Processor è contemporaneamente un subscriber e un publisher. Può ricevere elementi, trasformarli e pubblicare il risultato verso altri componenti.

public interface Processor<T, R>
        extends Subscriber<T>,
                Publisher<R> {
}

📡 Il ciclo di vita di un flusso reattivo

Dopo la sottoscrizione, un publisher può inviare una sequenza di segnali ben definita:

onSubscribe
     │
     ├── onNext
     ├── onNext
     ├── onNext
     │
     └── onComplete

oppure:

onSubscribe
     │
     ├── onNext
     ├── onNext
     │
     └── onError

La struttura generale è:

onNext × 0..N seguito da onComplete oppure onError

onComplete e onError sono segnali terminali. Dopo uno di questi eventi il publisher non deve inviare altri elementi.

🚦 Come funziona la backpressure

La backpressure risolve un problema fondamentale: che cosa accade quando il produttore genera dati più rapidamente di quanto il consumatore possa elaborarli?

Supponiamo che un publisher produca 10.000 eventi al secondo, mentre il subscriber sia in grado di processarne soltanto 500.

Senza controllo del flusso, le alternative principali sarebbero:

• accumulare continuamente gli eventi in memoria;
• bloccare il produttore;
• eliminare gli eventi in eccesso;
• provocare un errore quando il buffer è pieno.

Con la backpressure, il subscriber comunica quanti elementi è pronto a gestire:

@Override
public void onSubscribe(
        Flow.Subscription subscription
) {
    this.subscription = subscription;

    /*
     * Chiediamo inizialmente un solo elemento.
     */
    subscription.request(1);
}

@Override
public void onNext(String item) {

    process(item);

    /*
     * Terminata l'elaborazione, chiediamo
     * l'elemento successivo.
     */
    subscription.request(1);
}

La domanda scorre quindi nella direzione opposta rispetto ai dati:

DATI:
Publisher ─────────────────────────► Subscriber

DOMANDA:
Publisher ◄───────────────────────── Subscriber
                  request(n)

Strategie in caso di sovraccarico

☕ Reactive Programming con la Flow API di Java

Da Java 9, le interfacce compatibili con il modello Reactive Streams sono disponibili nella classe java.util.concurrent.Flow.

Le interfacce principali sono:

• Flow.Publisher<T>;
• Flow.Subscriber<T>;
• Flow.Subscription;
• Flow.Processor<T, R>.

La Flow API definisce il protocollo di comunicazione, ma non offre un insieme completo di operatori di trasformazione paragonabile a Reactor o RxJava.

È quindi particolarmente utile per:

• definire API interoperabili;
• creare publisher o subscriber semplici;
• comprendere il funzionamento dei Reactive Streams;
• integrare componenti che utilizzano le interfacce standard del JDK.

📤 Esempio con SubmissionPublisher

Java include SubmissionPublisher, un’implementazione di Flow.Publisher che permette di inviare elementi in modo asincrono ai subscriber.

Creare un subscriber

package it.esempio.reactive;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Flow;

public final class NumberSubscriber
        implements Flow.Subscriber<Integer> {

    private final CountDownLatch completionLatch;

    private Flow.Subscription subscription;

    public NumberSubscriber(
            CountDownLatch completionLatch
    ) {
        this.completionLatch = completionLatch;
    }

    @Override
    public void onSubscribe(
            Flow.Subscription subscription
    ) {
        this.subscription = subscription;

        System.out.println(
                "Sottoscrizione completata"
        );

        /*
         * Richiediamo un elemento alla volta.
         * Questo rende visibile il meccanismo
         * della backpressure.
         */
        subscription.request(1);
    }

    @Override
    public void onNext(Integer item) {

        System.out.println(
                "Elemento ricevuto: " + item
        );

        try {
            /*
             * Simuliamo un'elaborazione lenta.
             */
            Thread.sleep(200);

        } catch (InterruptedException exception) {
            Thread.currentThread().interrupt();

            subscription.cancel();
            completionLatch.countDown();

            return;
        }

        /*
         * Chiediamo il prossimo elemento solo
         * dopo aver completato quello corrente.
         */
        subscription.request(1);
    }

    @Override
    public void onError(Throwable throwable) {

        System.err.println(
                "Errore nel flusso: "
                + throwable.getMessage()
        );

        completionLatch.countDown();
    }

    @Override
    public void onComplete() {

        System.out.println(
                "Flusso completato"
        );

        completionLatch.countDown();
    }
}

Creare il publisher

package it.esempio.reactive;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.SubmissionPublisher;

public class FlowApplication {

    public static void main(String[] args)
            throws InterruptedException {

        CountDownLatch completionLatch =
                new CountDownLatch(1);

        /*
         * SubmissionPublisher deve essere chiuso
         * al termine dell'utilizzo.
         */
        try (
            SubmissionPublisher<Integer> publisher =
                    new SubmissionPublisher<>()
        ) {
            NumberSubscriber subscriber =
                    new NumberSubscriber(
                            completionLatch
                    );

            publisher.subscribe(subscriber);

            for (int number = 1; number <= 5; number++) {

                System.out.println(
                        "Pubblicazione: " + number
                );

                /*
                 * submit() invia l'elemento
                 * ai subscriber registrati.
                 */
                publisher.submit(number);
            }

            /*
             * close() segnala che non verranno
             * pubblicati altri elementi.
             */
            publisher.close();

            completionLatch.await();
        }
    }
}

Il subscriber chiede un elemento per volta. Dopo averlo elaborato, chiama nuovamente request(1).

🛠️ Principali librerie reattive per Java

⚛️ Introduzione a Project Reactor

Project Reactor è una libreria reattiva per la JVM basata sulla specifica Reactive Streams. Offre un ampio insieme di operatori per creare, trasformare, combinare e consumare flussi asincroni.

I suoi due tipi principali sono:

• Flux<T>: rappresenta una sequenza da zero a molti elementi;
• Mono<T>: rappresenta zero oppure un elemento.

Dipendenza Maven

<dependencies>

    <!--
        Libreria principale di Project Reactor.
        Verificare sempre la versione stabile compatibile
        con il proprio progetto.
    -->
    <dependency>
        <groupId>io.projectreactor</groupId>
        <artifactId>reactor-core</artifactId>
        <version>3.8.6</version>
    </dependency>

    <!--
        Strumenti per il test dei publisher reattivi.
    -->
    <dependency>
        <groupId>io.projectreactor</groupId>
        <artifactId>reactor-test</artifactId>
        <version>3.8.6</version>
        <scope>test</scope>
    </dependency>

</dependencies>

🔢 Differenza tra Flux e Mono

Creare un Mono

Mono<String> singleValue =
        Mono.just("Java");

Mono<String> emptyValue =
        Mono.empty();

Mono<String> failedValue =
        Mono.error(
                new IllegalStateException(
                        "Valore non disponibile"
                )
        );

Creare un Flux

Flux<String> languages =
        Flux.just(
                "Java",
                "Kotlin",
                "Scala"
        );

Flux<Integer> numbers =
        Flux.range(1, 10);

Flux<String> fromList =
        Flux.fromIterable(
                List.of(
                        "Spring",
                        "Reactor",
                        "WebFlux"
                )
        );

Creazione differita

Mono<Instant> currentTime =
        Mono.defer(
                () -> Mono.just(
                        Instant.now()
                )
        );

Mono.defer() esegue il supplier per ogni nuova sottoscrizione. È utile quando il valore deve essere calcolato al momento dell’utilizzo e non durante la costruzione della pipeline.

🔧 Gli operatori principali di Reactor

Gli operatori permettono di costruire una pipeline dichiarativa. Ogni operatore restituisce normalmente un nuovo publisher senza modificare quello originale.

Flux<String> result =
        Flux.range(1, 10)
                /*
                 * Conserviamo soltanto i numeri pari.
                 */
                .filter(number -> number % 2 == 0)

                /*
                 * Trasformiamo ogni numero in una stringa.
                 */
                .map(number -> "Numero " + number)

                /*
                 * Eseguiamo un'azione senza modificare
                 * l'elemento che attraversa il flusso.
                 */
                .doOnNext(value ->
                        System.out.println(
                                "Elaborazione: " + value
                        )
                );

result.subscribe(
        value -> System.out.println(
                "Ricevuto: " + value
        )
);

Operatori comuni

🗺️ Differenza tra map e flatMap

map deve essere utilizzato quando la funzione restituisce un normale valore.

Mono<Product> product =
        productService.findById(10L);

Mono<String> productName =
        product.map(Product::name);

flatMap deve essere utilizzato quando la funzione restituisce un altro publisher.

Mono<Product> product =
        productService.findById(10L);

Mono<ProductDetails> details =
        product.flatMap(
                value ->
                        detailsService.findByProductId(
                                value.id()
                        )
        );

Utilizzando map nel secondo esempio otterremmo un publisher annidato:

/*
 * Risultato poco pratico:
 * Mono<Mono<ProductDetails>>
 */
Mono<Mono<ProductDetails>> nested =
        product.map(
                value ->
                        detailsService.findByProductId(
                                value.id()
                        )
        );

flatMap con un Flux

Flux<Long> productIds =
        Flux.just(10L, 20L, 30L);

Flux<Product> products =
        productIds.flatMap(
                productService::findById
        );

flatMap può elaborare più publisher in concorrenza e non garantisce necessariamente l’ordine originale.

Quando l’ordine deve essere conservato, è possibile utilizzare concatMap:

Flux<Product> orderedProducts =
        productIds.concatMap(
                productService::findById
        );

🧬 Combinare più publisher

Una delle caratteristiche più utili della programmazione reattiva è la possibilità di combinare operazioni asincrone senza callback annidate.

Combinare due Mono con zip

Mono<Product> product =
        productService.findById(10L);

Mono<Stock> stock =
        stockService.findByProductId(10L);

Mono<ProductResponse> response =
        Mono.zip(product, stock)
                .map(tuple -> {

                    Product productValue =
                            tuple.getT1();

                    Stock stockValue =
                            tuple.getT2();

                    return new ProductResponse(
                            productValue.id(),
                            productValue.name(),
                            stockValue.availableUnits()
                    );
                });

Unire più Flux

Flux<String> first =
        Flux.just("A", "B");

Flux<String> second =
        Flux.just("C", "D");

Flux<String> merged =
        Flux.merge(first, second);

Flux<String> concatenated =
        Flux.concat(first, second);

merge combina gli elementi quando vengono prodotti, mentre concat completa il primo publisher prima di sottoscrivere il secondo.

Fallback per un publisher vuoto

Mono<Product> product =
        productService.findById(100L)
                .switchIfEmpty(
                        Mono.defer(
                                () ->
                                        productService
                                                .findDefaultProduct()
                        )
                );

▶️ Lazy execution e subscribe()

La maggior parte dei publisher Reactor è lazy: la costruzione della pipeline non avvia automaticamente l’elaborazione.

Flux<Integer> numbers =
        Flux.range(1, 5)
                .doOnNext(number ->
                        System.out.println(
                                "Elaborazione " + number
                        )
                );

/*
 * Fino a questo punto abbiamo soltanto
 * descritto la pipeline.
 */

/*
 * La sottoscrizione avvia il flusso.
 */
numbers.subscribe();

Ogni sottoscrizione può provocare una nuova esecuzione del publisher:

Mono<String> generated =
        Mono.fromSupplier(() -> {

            String value =
                    UUID.randomUUID().toString();

            System.out.println(
                    "Generato: " + value
            );

            return value;
        });

generated.subscribe();
generated.subscribe();

In questo caso il supplier viene eseguito due volte.

Le varianti di subscribe

Flux<Integer> numbers =
        Flux.range(1, 3);

numbers.subscribe(
        /*
         * Gestione del prossimo elemento.
         */
        value -> System.out.println(
                "Valore: " + value
        ),

        /*
         * Gestione dell'errore.
         */
        error -> System.err.println(
                "Errore: " + error.getMessage()
        ),

        /*
         * Gestione del completamento.
         */
        () -> System.out.println(
                "Completato"
        )
);

🧵 Thread e Scheduler in Project Reactor

Reactor non crea automaticamente un nuovo thread per ogni operatore. In assenza di istruzioni specifiche, la pipeline continua normalmente sul thread che ha effettuato la sottoscrizione.

Gli Scheduler permettono di controllare dove vengono eseguite determinate parti della catena.

Scheduler principali

Differenza tra subscribeOn e publishOn

Flux<Integer> numbers =
        Flux.range(1, 5)

                /*
                 * Influenza il contesto in cui avviene
                 * la sottoscrizione alla sorgente.
                 */
                .subscribeOn(
                        Schedulers.boundedElastic()
                )

                .map(number -> number * 2)

                /*
                 * Gli operatori successivi vengono
                 * eseguiti sullo Scheduler indicato.
                 */
                .publishOn(
                        Schedulers.parallel()
                )

                .map(number -> number + 1);

subscribeOn influenza il punto in cui viene sottoscritta la sorgente, mentre publishOn cambia il contesto di esecuzione degli operatori successivi.

⛔ Integrare codice bloccante

Il principale vantaggio di uno stack reattivo non bloccante viene perso quando un’operazione lenta blocca un thread dell’event loop.

Questo esempio è pericoloso in un controller WebFlux:

public Mono<String> readFile(Path path) {

    /*
     * Files.readString() è un'operazione bloccante.
     * Non deve essere eseguita direttamente
     * sul thread reattivo dell'event loop.
     */
    return Mono.just(
            Files.readString(path)
    );
}

Inoltre, Mono.just() valuta l’argomento immediatamente. Una possibile integrazione controllata è:

public Mono<String> readFile(Path path) {

    return Mono.fromCallable(
                    /*
                     * L'operazione viene eseguita
                     * al momento della sottoscrizione.
                     */
                    () -> Files.readString(path)
            )
            /*
             * boundedElastic è progettato per isolare
             * operazioni bloccanti con limiti controllati.
             */
            .subscribeOn(
                    Schedulers.boundedElastic()
            );
}

Evitare block() nella pipeline applicativa

/*
 * Da evitare all'interno di una catena WebFlux:
 * il thread resta in attesa del risultato.
 */
Product product =
        productService.findById(10L)
                .block();

block() può essere utile in strumenti da riga di comando, test o punti di integrazione controllati, ma non dovrebbe essere usato casualmente nel percorso reattivo di una richiesta.

🚨 Gestire gli errori in Reactor

In un flusso reattivo, un errore è un segnale terminale. Quando viene emesso onError, la sequenza originale termina.

Restituire un valore predefinito

Mono<Product> product =
        productService.findById(10L)
                .onErrorReturn(
                        new Product(
                                0L,
                                "Prodotto temporaneamente non disponibile"
                        )
                );

Utilizzare un publisher alternativo

Mono<Product> product =
        productService.findById(10L)
                .onErrorResume(
                        DatabaseException.class,
                        error -> {

                            /*
                             * In caso di errore sul database,
                             * recuperiamo il prodotto dalla cache.
                             */
                            return cacheService
                                    .findProduct(10L);
                        }
                );

Trasformare l’eccezione

Mono<Product> product =
        productService.findById(10L)
                .onErrorMap(
                        DatabaseException.class,
                        error ->
                                new ProductServiceException(
                                        "Impossibile recuperare il prodotto",
                                        error
                                )
                );

Eseguire logging senza sostituire l’errore

Mono<Product> product =
        productService.findById(10L)
                .doOnError(error ->
                        logger.error(
                                "Errore durante il recupero",
                                error
                        )
                );

Eseguire una pulizia al termine

Mono<Product> product =
        productService.findById(10L)
                .doFinally(signalType -> {

                    /*
                     * Viene eseguito in caso di successo,
                     * errore o cancellazione.
                     */
                    logger.info(
                            "Flusso terminato: {}",
                            signalType
                    );
                });

🔁 Retry, timeout e operazioni remote

Il retry può essere utile per errori temporanei, ma deve essere utilizzato con attenzione.

Retry semplice

Mono<Product> product =
        remoteProductService.findById(10L)
                /*
                 * Dopo l'errore, la sorgente
                 * viene sottoscritta nuovamente.
                 */
                .retry(2);

Retry con attesa progressiva

Mono<Product> product =
        remoteProductService.findById(10L)
                .retryWhen(
                        Retry.backoff(
                                        3,
                                        Duration.ofMillis(200)
                                )
                                /*
                                 * Ripetiamo soltanto
                                 * gli errori temporanei.
                                 */
                                .filter(
                                        TemporaryServiceException.class
                                                ::isInstance
                                )
                );

Aggiungere un timeout

Mono<Product> product =
        remoteProductService.findById(10L)
                .timeout(
                        Duration.ofSeconds(2)
                )
                .onErrorResume(
                        TimeoutException.class,
                        error ->
                                cacheService
                                        .findProduct(10L)
                );

🔥 Publisher cold e hot

Un publisher cold produce normalmente una nuova sequenza per ogni subscriber.

Flux<Integer> coldFlux =
        Flux.range(1, 3)
                .doOnSubscribe(subscription ->
                        System.out.println(
                                "Nuova sottoscrizione"
                        )
                );

coldFlux.subscribe(
        value -> System.out.println(
                "Subscriber A: " + value
        )
);

coldFlux.subscribe(
        value -> System.out.println(
                "Subscriber B: " + value
        )
);

Entrambi i subscriber ricevono l’intera sequenza.

Un publisher hot può invece produrre elementi indipendentemente dalla presenza di un subscriber. Chi si iscrive in ritardo potrebbe non ricevere gli eventi precedenti.

Creare un flusso hot con Sinks

Sinks.Many<String> sink =
        Sinks.many()
                .multicast()
                .onBackpressureBuffer();

Flux<String> notifications =
        sink.asFlux();

notifications.subscribe(
        value -> System.out.println(
                "Utente A: " + value
        )
);

sink.tryEmitNext(
        "Prima notifica"
);

notifications.subscribe(
        value -> System.out.println(
                "Utente B: " + value
        )
);

sink.tryEmitNext(
        "Seconda notifica"
);

L’utente B non riceve necessariamente la prima notifica perché si è collegato successivamente.

Condividere un publisher cold

Flux<Long> shared =
        Flux.interval(
                    Duration.ofSeconds(1)
                )
                .share();

share() permette a più subscriber di condividere la stessa sottoscrizione alla sorgente.

🚥 Backpressure con Reactor

Reactor offre diversi operatori per definire il comportamento in caso di pressione eccessiva.

Bufferizzare gli elementi

Flux<Event> buffered =
        eventSource()
                .onBackpressureBuffer(
                        1_000
                );

Scartare gli elementi in eccesso

Flux<Event> dropping =
        eventSource()
                .onBackpressureDrop(
                        droppedEvent ->
                                logger.warn(
                                        "Evento scartato: {}",
                                        droppedEvent
                                )
                );

Mantenere soltanto l’ultimo elemento

Flux<SensorValue> latest =
        sensorValues()
                .onBackpressureLatest();

Controllare la richiesta durante subscribe

Flux.range(1, 100)
        .subscribe(
                new BaseSubscriber<Integer>() {

                    @Override
                    protected void hookOnSubscribe(
                            Subscription subscription
                    ) {
                        /*
                         * Richiediamo il primo elemento.
                         */
                        request(1);
                    }

                    @Override
                    protected void hookOnNext(
                            Integer value
                    ) {
                        System.out.println(
                                "Elaborazione: " + value
                        );

                        /*
                         * Terminato il lavoro, chiediamo
                         * un altro elemento.
                         */
                        request(1);
                    }
                }
        );

🧪 Testare una pipeline con StepVerifier

StepVerifier permette di descrivere i segnali attesi da un publisher e verificare che vengano emessi nell’ordine corretto.

Testare un Flux

import org.junit.jupiter.api.Test;
import reactor.core.publisher.Flux;
import reactor.test.StepVerifier;

class ProductServiceTest {

    @Test
    void shouldReturnActiveProductNames() {

        Flux<String> names =
                Flux.just(
                                new Product(
                                        1L,
                                        "Laptop",
                                        true
                                ),
                                new Product(
                                        2L,
                                        "Monitor",
                                        false
                                ),
                                new Product(
                                        3L,
                                        "Tastiera",
                                        true
                                )
                        )
                        .filter(Product::active)
                        .map(Product::name);

        StepVerifier.create(names)
                .expectNext(
                        "Laptop",
                        "Tastiera"
                )
                .verifyComplete();
    }
}

Testare un errore

@Test
void shouldEmitAnError() {

    Mono<Product> result =
            Mono.error(
                    new ProductNotFoundException(
                            "Prodotto non trovato"
                    )
            );

    StepVerifier.create(result)
            .expectErrorMatches(
                    error ->
                            error instanceof
                                    ProductNotFoundException
                            &&
                            error.getMessage()
                                    .equals(
                                            "Prodotto non trovato"
                                    )
            )
            .verify();
}

Testare il tempo virtuale

@Test
void shouldEmitAfterDelay() {

    StepVerifier.withVirtualTime(
                    () ->
                            Mono.delay(
                                    Duration.ofHours(1)
                            )
                )
                .expectSubscription()
                .expectNoEvent(
                        Duration.ofHours(1)
                )
                .expectNext(0L)
                .verifyComplete();
}

Il tempo virtuale permette di testare ritardi lunghi senza attendere realmente un’ora.

🌐 Creare una API reattiva con Spring WebFlux

Spring WebFlux è il framework web reattivo dello Spring Framework. Supporta Reactive Streams e può funzionare su server non bloccanti come Reactor Netty.

Dipendenza Spring Boot

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-webflux</artifactId>
</dependency>

Definire il modello

package it.esempio.webflux;

import java.math.BigDecimal;

public record Product(
        Long id,
        String name,
        BigDecimal price
) {
}

Creare il servizio

package it.esempio.webflux;

import org.springframework.stereotype.Service;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.math.BigDecimal;

@Service
public class ProductService {

    public Flux<Product> findAll() {

        return Flux.just(
                new Product(
                        1L,
                        "Laptop",
                        new BigDecimal("899.00")
                ),
                new Product(
                        2L,
                        "Monitor",
                        new BigDecimal("249.00")
                ),
                new Product(
                        3L,
                        "Tastiera",
                        new BigDecimal("79.90")
                )
        );
    }

    public Mono<Product> findById(Long id) {

        return findAll()
                .filter(product ->
                        product.id().equals(id)
                )
                .next();
    }
}

Creare il controller

package it.esempio.webflux;

import org.springframework.http.HttpStatus;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.server.ResponseStatusException;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

@RestController
@RequestMapping("/api/products")
public class ProductController {

    private final ProductService productService;

    public ProductController(
            ProductService productService
    ) {
        this.productService = productService;
    }

    @GetMapping
    public Flux<Product> findAll() {

        /*
         * Non chiamiamo subscribe():
         * Spring WebFlux gestisce la sottoscrizione.
         */
        return productService.findAll();
    }

    @GetMapping("/{id}")
    public Mono<Product> findById(
            @PathVariable Long id
    ) {
        return productService.findById(id)
                .switchIfEmpty(
                        Mono.error(
                                new ResponseStatusException(
                                        HttpStatus.NOT_FOUND,
                                        "Prodotto non trovato"
                                )
                        )
                );
    }
}

Streaming con Server-Sent Events

package it.esempio.webflux;

import org.springframework.http.MediaType;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;
import reactor.core.publisher.Flux;

import java.time.Duration;
import java.time.Instant;

@RestController
public class EventController {

    @GetMapping(
        value = "/api/events",
        produces = MediaType.TEXT_EVENT_STREAM_VALUE
    )
    public Flux<SystemEvent> events() {

        return Flux.interval(
                    Duration.ofSeconds(1)
                )
                .map(sequence ->
                        new SystemEvent(
                                sequence,
                                Instant.now(),
                                "Sistema operativo"
                        )
                );
    }
}

Il client riceve gli eventi progressivamente, senza attendere il completamento dell’intera sequenza.

Utilizzare WebClient

package it.esempio.webflux;

import org.springframework.stereotype.Service;
import org.springframework.web.reactive.function.client.WebClient;
import reactor.core.publisher.Mono;

@Service
public class RemoteProductClient {

    private final WebClient webClient;

    public RemoteProductClient(
            WebClient.Builder builder
    ) {
        this.webClient = builder
                .baseUrl(
                        "https://api.example.com"
                )
                .build();
    }

    public Mono<Product> findById(Long id) {

        return webClient.get()
                .uri(
                        "/products/{id}",
                        id
                )
                .retrieve()
                .bodyToMono(Product.class);
    }
}

🧯 Errori comuni nella programmazione reattiva

Dimenticare di restituire il publisher modificato

Gli operatori restituiscono un nuovo publisher. Non modificano necessariamente l’istanza precedente.

Flux<String> names =
        Flux.just(
                "Java",
                "Reactor",
                "Spring"
        );

/*
 * Il risultato di map viene ignorato.
 */
names.map(String::toUpperCase);

/*
 * Questa sottoscrizione riceve ancora
 * i valori originali.
 */
names.subscribe(System.out::println);

La versione corretta è:

Flux<String> upperCaseNames =
        names.map(
                String::toUpperCase
        );

upperCaseNames.subscribe(
        System.out::println
);

🎯 Quando utilizzare la Reactive Programming

La programmazione reattiva è particolarmente adatta alle applicazioni con molte operazioni concorrenti e una forte componente di input/output.

Casi adatti

• API che devono gestire molte connessioni contemporanee;
• gateway e servizi di aggregazione;
• chiamate verso numerosi servizi remoti;
• streaming di dati;
• Server-Sent Events e WebSocket;
• applicazioni event-driven;
• microservizi con client e database non bloccanti;
• elaborazione di flussi potenzialmente infiniti;
• sistemi che richiedono backpressure;
• dashboard e notifiche in tempo reale.

Quando potrebbe non essere necessaria

• applicazioni CRUD semplici con un carico limitato;
• progetti interamente basati su librerie bloccanti;
• elaborazioni prevalentemente CPU-bound;
• batch sequenziali di piccole dimensioni;
• team senza esperienza reattiva e senza un reale problema di scalabilità;
• applicazioni nelle quali la semplicità è più importante della concorrenza elevata.

✅ Best practice

1. Mantenere la pipeline non bloccante

Una sola chiamata bloccante nel punto sbagliato può rallentare molte richieste gestite dallo stesso event loop.

2. Restituire publisher dai servizi

public Mono<Product> findProduct(
        long id
) {
    return repository.findById(id)
            .flatMap(
                    enrichmentService::enrich
            );
}

3. Gestire sempre gli errori attesi

Timeout, indisponibilità temporanee e risultati vuoti devono essere parte del flusso applicativo e non situazioni ignorate.

4. Applicare limiti alla concorrenza

Flux<ProductDetails> details =
        productIds.flatMap(
                detailsService::findById,

                /*
                 * Numero massimo di operazioni
                 * concorrenti.
                 */
                8
        );

5. Scegliere consapevolmente tra flatMap e concatMap

La concorrenza è utile, ma l’ordine può essere importante per determinati flussi.

6. Non utilizzare buffer illimitati

Un buffer senza limiti può nascondere temporaneamente il problema del consumatore lento, fino a causare un consumo eccessivo di memoria.

7. Definire timeout per le risorse remote

Una chiamata che non termina può trattenere risorse e rallentare l’intera pipeline.

8. Testare completamento, errori e cancellazione

Un test reattivo non deve verificare soltanto i valori, ma anche l’ordine dei segnali e la conclusione del flusso.

9. Monitorare la pipeline

È utile raccogliere almeno:

• tempo di risposta;
• numero di sottoscrizioni;
• errori per tipologia;
• retry eseguiti;
• timeout;
• elementi scartati;
• dimensione dei buffer;
• latenza dei servizi esterni.

10. Evitare pipeline eccessivamente lunghe

Una catena con decine di operatori può diventare difficile da leggere. È preferibile estrarre parti significative in metodi con nomi descrittivi.

public Flux<ProductResponse> findAvailableProducts() {

    return repository.findAll()
            .filter(Product::active)
            .transform(
                    this::enrichWithStock
            )
            .map(
                    productMapper::toResponse
            );
}

private Flux<ProductWithStock> enrichWithStock(
        Flux<Product> products
) {
    return products.flatMap(
            product ->
                    stockService
                            .findByProductId(
                                    product.id()
                            )
                            .map(stock ->
                                    new ProductWithStock(
                                            product,
                                            stock
                                    )
                            )
    );
}

🏁 Conclusioni

La Reactive Programming in Java permette di rappresentare dati ed eventi come flussi asincroni componibili. Il produttore pubblica gli elementi, il consumatore reagisce ai segnali e la backpressure permette di controllare la quantità di dati richiesta.

La specifica Reactive Streams definisce le interfacce fondamentali Publisher, Subscriber, Subscription e Processor. Java mette a disposizione le rispettive interfacce nella classe java.util.concurrent.Flow.

Project Reactor aggiunge un modello più ricco basato su Flux e Mono, insieme a operatori per trasformazione, combinazione, gestione degli errori, scheduling, retry e controllo della pressione.

Spring WebFlux utilizza questi concetti per costruire applicazioni web non bloccanti e capaci di gestire numerose operazioni di input/output concorrenti.

La programmazione reattiva non deve però essere considerata una soluzione universale. Richiede una maggiore disciplina, una corretta gestione degli errori, conoscenza dei thread e attenzione alle chiamate bloccanti. Il suo utilizzo è giustificato soprattutto quando l’applicazione deve elaborare flussi continui o gestire molte attese concorrenti con un numero controllato di risorse.

📖 Documentazione ufficiale

• 
  Reactive Streams
  
• 
  Java API: java.util.concurrent.Flow
  
• 
  Java API: SubmissionPublisher
  
• 
  Project Reactor Reference Guide
  
• 
  Reactor Testing e StepVerifier
  
• 
  Documentazione Spring WebFlux
  
• 
  Progetto ufficiale RxJava
  
• 
  The Reactive Manifesto
  

“`