Introducción a Spring Reactor y Spring Webflux

Introducción

Este capítulo presenta brevemente las novedades responsivas de Spring 5.

El objetivo de la programación responsiva es simplificar la escritura de aplicaciones asíncronas que, en lugar de utilizar el mecanismo clásico de un pool voluminoso de threads bloqueantes, utiliza un bucle y un número mínimo de threads gracias a un motor basado en un framework NIO como Netty.

Ponemos en cola los procesamientos rápidos de realizar, y se llevan a cabo a lo largo del tiempo. Entonces, es posible modelar estos procesamientos en forma de flujos y propagaciones de cambios. Podemos gestionar flujos estáticos con tablas o flujos dinámicos con transmisores y receptores de eventos, como en el design pattern observador.

En un contexto de servidor de un clúster elástico, este tipo de procesamiento «en serie» permite saber si la carga en un servidor está alineada con la potencia del clúster a través de la backpressure, que mide el volumen del flujo aceptable en la entrada del servidor. Si se supera un umbral, podemos aumentar el número activo de nodos en el clúster para aligerar la carga por servidor y, seguidamente, reducir el número de nodos si la carga baja.

El objetivo es que los productores (Publisher) no abrumen a los consumidores (Consumer). Esto es particularmente adecuado para microservicios en el cloud.

El diseño de servidores...

Spring Reactor

WebFlux

Normalmente usaremos Reactor como parte de las aplicaciones WebFlux.

De hecho, las API Responsive en flujos de eventos estáticos a menudo se sustituyen por Streams tradicionales. WebFlux es el equivalente responsivo de Spring MVC y utiliza el motor Reactor internamente.

Diseñar una aplicación WebFlux desde cero es complejo. Recientemente, JHipster permite generar aplicaciones completas usándolo, y su uso puede ahorrar mucho tiempo.

Para utilizar Spring WebFlux, debe añadir la dependencia Maven:

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

Hay disponible un ejemplo completo generado con JHipster en los ejemplos descargables del libro. Los detalles al respecto se estudian en un capítulo específico.

Pruebas con WebFlux

Spring ofrece una serie de herramientas para facilitar las pruebas.

public class UsuarioService { 
 public Mono<Usuario> getUsuarioById(Integer usuarioId) { 
 return webClient 
   .get() 
   .uri("http://localhost:8080/usuario/{id}", usuarioId) 
   .retrieve() 
   .bodyToMono(Usuario.class); 
 } 
} 

@ExtendWith(MockitoExtension.class) 
public class UsuarioServiceTest { 
 
 @Test 
 void givenUsuarioId_whenGetUsuarioById_thenReturnUsuario() { 
 
 Integer usuarioId = 100; 
 Usuario mockUsuario = new Usuario(1, "Pepe", 51); 
 when(webClientMock.get()) 
   .thenReturn(requestHeadersUriSpecMock); 
 when(requestHeadersUriMock.uri("/usuario/{id}", usuarioId)) 
   .thenReturn(requestHeadersSpecMock); 
 when(requestHeadersMock.retrieve()) 
   .thenReturn(responseSpecMock); 
 when(responseMock.bodyToMono(Usuario.class)) 
   .thenReturn(Mono.just(mockUsuario)); 
 
 Mono<Usuario> usuarioMono =  
usuarioService.getUsuarioById(usuarioId); 
 
 StepVerifier.create(usuarioMono) 
   .expectNextMatches(usuario -> usuario.getNombre() 
   .equals("Pepe")) 
   .verifyComplete(); 
 } 
} 

Server Site Event con Spring

En nuestras aplicaciones, a menudo se requieren refrescos para llamadas asíncronas. Hay varias tecnologías para esto. La más sencilla es el pooling, que consiste en consultar regularmente al servidor para conocer el avance de un procesamiento. Otra evolución es el long pooling, que consiste en mantener abierta la conexión HTTP; pero estas dos técnicas son costosas a nivel de red. Existe una evolución con webhooks y Publish-Subscribe. También tenemos los websockets, que son bidireccionales, y Server Site Event o SSE, para abreviar, que es un estándar HTTP que permite a una aplicación web manejar un flujo unidireccional de eventos y recibir actualizaciones cada vez que el servidor emite datos. Esta solución suele ser más adecuada en un contexto web que no necesite la característica en tiempo real de los websockets. Por lo tanto, perdemos algo de tiempo, pero algunas veces esto es sostenible.

La versión Spring 4.2 ya soportaba SSE, pero, a partir de Spring 5, hay una forma más idiomática y práctica de gestionarlo a través de un flujo y un elemento ServerSiteEvent, gracias a las API responsivas de Spring WebFlux. Creamos un endpoint de streaming SSE, siguiendo las especificaciones del W3C que designan el tipo MIME como text/event-stream:

@GetMapping(path = "/stream-flux"...