Creando una API reactiva con RxJava y Micronaut

Índice de contenidos

1. Introducción
2. Entorno
3. Programación reactiva con Micronaut
4. Implementación de API reactiva con RxJava

1. Introducción

En el tutorial anterior, «Creando mi primera aplicación con Micronaut», vimos las características principales de este nuevo framework y lo sencillo que era crear nuestras aplicaciones basadas en microservicios.

En este tutorial nos vamos a centrar en cómo Micronaut permite crear nuestras APIs de manera reactiva consiguiendo así reducir los tiempos de espera, funcionamiento óptimo ante cargas elevadas en el sistema o  más robustez ante fallos o caídas en nuestras aplicaciones.

2. Entorno

• Hardware: Portátil MacBook Pro 15 pulgadas (2.5 GHz Intel i7, 16GB 1600 Mhz DDR3, 500GB Flash Storage).
• Sistema Operativo: MacOs Mojave 10.14.2
• Micronaut 1.0.2

3. Programación reactiva con Micronaut

Micronaut está implementado sobre Netty, que es un framework basado en eventos y un modelo de entrada/salida no bloqueante. Gracias a ello, implementar nuestras APIs reactivas es muy sencillo, dejando el control al framework para decidir si una petición debe tratarse de manera síncrona o asíncrona.

A pesar de estar construído sobre Netty, Micronaut da la opción de diseñar APIs bloqueantes, ya que existen escenarios donde las respuestas deben servirse de manera síncrona. Es responsabilidad de nosotros decidir qué tipo de API diseñar para cada escenario.

¿Cómo sabe Micronaut si la petición es bloqueante o no? En función del tipo de respuesta del servicio:

• Si el método del controlador devuelve un tipo no bloqueante como por ejemplo CompletableFuture , Observable de RxJava o Mono de Reactor, la petición se servirá de manera asíncrona, siendo tratada por el event loop de Netty.
• Si el método del controlador devuelve otro tipo de dato (p.e. String) la petición será tratada por el pool de hilos creado al iniciarse la aplicación y que es bloqueante. Este pool de hilos puede configurarse al igual que con otros frameworks como Spring Boot.

En el siguiente apartado veremos una implementación de dos puntos de API uno bloqueante y otro no bloqueante.

4. Implementación de una API reactiva con RxJava

Creamos un nuevo controlador, EchoController, que contiene dos puntos de API que devuelven como respuesta el contenido del body de la petición:

@Controller("/echo")
public class EchoController {

    @Post(value = "/sync", consumes = MediaType.TEXT_PLAIN)
    public String echoSync(@Size(max = 1024) @Body String text) {
        return text;
    }

    @Post(value = "/async", consumes = MediaType.TEXT_PLAIN)
    public Single<MutableHttpResponse<String>> echoAsync(@Body Flowable<String> text) {
        return text.collect(StringBuffer::new, StringBuffer::append)
                   .map(buffer ->
                           HttpResponse.ok(buffer.toString())
                   );
    }
}

1. El método echoSync define un tamaño máximo de 1MB para la petición, almacenando en memoria el contenido de la petición. Para peticiones con muchos datos, esto puede provocar fallos de memoria.
2. El método echoAsync soluciona este problema, ya que va almacenando cada fragmento del body en el buffer y respondiendo cuando termina de procesar el contenido de text sin ser una petición bloqueante gracias al tipo de dato Single de RxJava.

Resumiendo, en función de los tipos de la petición y respuesta de nuestros métodos, Micronaut sabe quién gestionará la llamada haciendo la petición bloqueante o no de manera muy sencilla, permitiendo configurar también el pool de hilos. Ahora es competencia de cada uno el saber cuándo utilizar una implementación u otra 😉