Pipeline 100% Reattive: Gestire flussi continui con Spring WebFlux, PostgreSQL, R2DBC e Docker

Quando si progetta un'API destinata all'elaborazione di stream continui di dati, l'approccio sincrono tradizionale basato su JDBC e Spring MVC mostra rapidamente i propri limiti architetturali, portando frequentemente alla saturazione dei thread a disposizione del server. L'adozione di Spring WebFlux in sinergia con R2DBC e PostgreSQL consente di superare tale ostacolo, implementando una pipeline end-to-end asincrona e non bloccante. Questo paradigma garantisce un flusso dati ottimizzato tra database e client, riducendo drasticamente l'impatto sulla memoria di sistema.

1. Configurazione dell'Infrastruttura Reattiva

L'implementazione del supporto reattivo richiede l'inclusione delle dipendenze relative a WebFlux, Spring Data R2DBC e il driver asincrono specifico per PostgreSQL all'interno del descrittore di progetto (Maven o Gradle).

<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-webflux</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-data-r2dbc</artifactId>
    </dependency>
    <dependency>
        <groupId>org.postgresql</groupId>
        <artifactId>r2dbc-postgresql</artifactId>
    </dependency>
</dependencies>

Nel file di configurazione application.yml, è imperativo sostituire il prefisso JDBC standard per inizializzare il connection pool reattivo. Di seguito una configurazione production-ready che include la gestione ottimizzata del pool di connessioni e il tracciamento delle query:

spring:
  r2dbc:
    url: ${DB_URL:r2dbc:postgresql://localhost:5432/telemetry_db}
    username: ${DB_USER:postgres}
    password: ${DB_PASSWORD:postgres}
    pool:
      initial-size: 10
      max-size: 50
      max-idle-time: 30m
      validation-query: "SELECT 1"
logging:
  level:
    org.springframework.r2dbc: DEBUG # Espone i log delle query reattive in fase di sviluppo

2. Modellazione del Dominio e Livello di Persistenza

Il framework Spring Data R2DBC si discosta dall'astrazione sincrona tipica di JPA e non utilizza l'Entity Manager. Il mapping oggetto-relazionale avviene tramite annotazioni native e dirette. La seguente classe rappresenta l'entità SensorMetric, strutturata sfruttando Lombok per minimizzare il codice boilerplate:

import org.springframework.data.annotation.Id;
import org.springframework.data.relational.core.mapping.Table;
import org.springframework.data.relational.core.mapping.Column;
import lombok.Data;
import lombok.NoArgsConstructor;
import lombok.AllArgsConstructor;
import java.time.LocalDateTime;

@Data
@NoArgsConstructor
@AllArgsConstructor
@Table("sensor_metrics")
public class SensorMetric {
    @Id
    private Long id;

    @Column("sensor_id")
    private String sensorId;

    private Double reading;

    @Column("recorded_at")
    private LocalDateTime timestamp;
}

Il livello di persistenza è demandato a un'interfaccia che estende ReactiveCrudRepository, esponendo i tipi reattivi nativi di Project Reactor (Flux per collezioni e Mono per risultati singoli):

import org.springframework.data.repository.reactive.ReactiveCrudRepository;
import org.springframework.stereotype.Repository;
import reactor.core.publisher.Flux;

@Repository
public interface SensorMetricRepository extends ReactiveCrudRepository<SensorMetric, Long> {
    // Il framework genera dinamicamente una query asincrona non bloccante
    Flux<SensorMetric> findBySensorId(String sensorId);
}

3. Implementazione dello Streaming nel Controller

A livello di Controller, l'impiego di TEXT_EVENT_STREAM_VALUE permette di instaurare una comunicazione Server-Sent Events (SSE) verso il client, mentre APPLICATION_NDJSON_VALUE risulta ottimale per l'ingestione efficiente di flussi continui senza caricare l'intero payload in RAM.

import org.springframework.http.HttpStatus;
import org.springframework.http.MediaType;
import org.springframework.web.bind.annotation.*;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

@RestController
@RequestMapping("/api/v1/metrics")
public class MetricStreamController {

    private final SensorMetricRepository repository;

    public MetricStreamController(SensorMetricRepository repository) {
        this.repository = repository;
    }

    @GetMapping(value = "/stream/{sensorId}", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public Flux<SensorMetric> streamMetricsOut(@PathVariable String sensorId) {
        return repository.findBySensorId(sensorId);
    }

    @PostMapping(value = "/ingest", consumes = MediaType.APPLICATION_NDJSON_VALUE)
    @ResponseStatus(HttpStatus.CREATED)
    public Mono<Void> ingestMetricsIn(@RequestBody Flux<SensorMetric> metricsStream) {
        return repository.saveAll(metricsStream).then();
    }
}

L'aspetto architetturale più rilevante di questa implementazione è il meccanismo di Backpressure. Qualora il client in lettura presenti latenze operative, il server Netty notifica Project Reactor affinché adegui il rate di emissione. Tale segnale viene propagato a cascata fino al driver R2DBC, che sospenderà temporaneamente le richieste di fetch al demone PostgreSQL, prevenendo attivamente l'esaurimento della memoria heap.

4. Orchestrazione Docker e Integrazione dei Dati Spaziali (PostGIS)

Nello sviluppo di moduli software complessi, quali sistemi per l'estrazione di supply plan o l'analisi di telemetria geolocalizzata in formato GeoJSON, l'architettura reattiva dimostra un'elevata scalabilità se integrata con l'estensione PostGIS. Il driver R2DBC per PostgreSQL supporta nativamente i tipi geometrici spaziali, consentendo di elaborare ed esportare grandi volumi di coordinate senza causare il blocco dei thread applicativi.

La configurazione docker-compose.yml seguente illustra come orchestrare agevolmente l'API insieme a un'istanza database dotata di estensione spaziale preconfigurata:

version: '3.8'
services:
  postgres-postgis:
    image: postgis/postgis:15-alpine
    environment:
      POSTGRES_DB: telemetry_db
      POSTGRES_USER: postgres
      POSTGRES_PASSWORD: postgres
    ports:
      - "5432:5432"

  webflux-api:
    build: .
    ports:
      - "8080:8080"
    environment:
      DB_URL: r2dbc:postgresql://postgres-postgis:5432/telemetry_db

5. Validazione End-to-End (E2E) in Pipeline CI/CD

Il collaudo di endpoint asincroni (SSE e NDJSON) evidenzia spesso i limiti strutturali dei client REST tradizionali. L'approccio ingegneristico più solido consiste nell'implementazione di una suite di test automatizzati tramite Karate DSL. Grazie alla sua sintassi dichiarativa, Karate permette di validare flussi di dati continui ed è facilmente integrabile in pipeline di Continuous Integration (es. Jenkins).

Feature: Validazione E2E dell'API WebFlux Stream

  Background:
    * url 'http://localhost:8080/api/v1/metrics'

  Scenario: Ingestione bulk in NDJSON e verifica stream in lettura SSE
    Given path 'ingest'
    And header Content-Type = 'application/x-ndjson'
    And request '{"sensorId": "S-100", "reading": 22.5}\n{"sensorId": "S-100", "reading": 23.1}'
    When method post
    Then status 201

    Given path 'stream/S-100'
    When method get
    Then status 200
    And match header Content-Type contains 'text/event-stream'

L'adozione sinergica di queste tecnologie definisce un ecosistema software resiliente, progettato per sostenere elevati carichi transazionali e picchi di traffico massicci, pur mantenendo un ridotto e controllato consumo delle risorse infrastrutturali.