Introducción

Los lenguajes de programación y los frameworks evolucionan con el tiempo para ofrecer a los desarrolladores una forma más simple y eficaz de resolver los problemas del mundo real y conseguir una solución adecuada según el contexto. Por ejemplo, cuando nació la programación orientada a objetos se pudo romper la barrera entre desarrollador y experto de producto plasmando la problemática en objetos de dominio que actuaban como lenguaje común para ambas partes.

Otra gran necesidad que surgió fue la de no utilizar el clásico modelo de programación «instrucción por instrucción», donde es necesario que para poder pasar a la siguiente instrucción, la anterior haya finalizado su propósito. En su lugar, se busca un planteamiento asíncrono que permita aprovechar al máximo los hilos de ejecución de nuestra aplicación para brindarle al usuario una experiencia más rápida y flexible. Es aquí donde aparece el concepto de Programación Reactiva.

Cabe destacar que se originó en la década de 1990 con el concepto de programación funcional reactiva. Científicos de la computación como David Pearce, Conal Elliott y Paul Hudak, trabajaron en el desarrollo de lenguajes y teorías que dieron lugar a la programación reactiva tal como la conocemos hoy en día.

En 2003, un grupo de desarrolladores liderados por Anders Hejlsberg en Microsoft, crearon Reactive Extensions (Rx), una biblioteca para programación reactiva en .NET. Fue ahí que comenzó a ganar mucha popularidad y fue siendo adoptada en muchos otros lenguajes y frameworks.

¿Qué es la programación reactiva?

La programación reactiva es un paradigma de programación que se centra en la gestión de datos que cambian a lo largo del tiempo. En lugar de tratar los datos como valores estáticos, se tratan como flujos de datos que cambian en el tiempo.

Un programa reactivo se basa en la idea de que los datos pueden ser tratados como un flujo continuo de eventos que se pueden procesar de forma asíncrona. Los programas reactivos pueden manejar grandes cantidades de datos y son adecuados para aplicaciones que necesitan procesar grandes cantidades de información de manera eficiente, como aplicaciones en tiempo real o aplicaciones que manejan mucho tráfico de red.

Un bloque de código se subscribe a un flujo de datos continuo definido por otro código para comenzar a recibir notificaciones (datos) cuando se producen cambios en los datos. Gracias a esto, nuestro código se vuelve asíncrono y no bloqueante, permitiendo que un hilo de ejecución no se bloquee por estar ejecutando una instrucción que depende de recibir una serie de datos y que delegue el trabajo a otro hilo para poder seguir procesando otras instrucciones, y este hilo a su vez no depende de recibir los datos en el momento, si no que es capaz de subscribirse a la entrada de esos datos para ir recibiéndolos a medida que están listos.

Además de la asincronía, otra gran ventaja que nos brinda este paradigma es la posibilidad de ser más resiliente a los fallos, o diferentes enfoques para evitarlos. Por ejemplo, si tenemos que consumir una gran cantidad de datos desde una fuente pero sabemos que si cargamos todos de golpe nuestro sistema se va a sobrecargar y por tanto dejará de funcionar, podemos poner límites ya sea por lotes en los que consumimos poco a poco todos los datos, o incluso controlando la carga por tiempo, cargando solo durante un tiempo y reanudando después. Hay muchas posibilidades disponibles para usar según sea el problema que tengamos que resolver.

Ejemplo (Spring Boot + Kotlin)

El código fuente de los siguientes ejemplos lo puedes encontrar alojado en Github visitando el siguiente enlace:

https://github.com/raulpadilladelgado/reactive-springboot

Publicar API Rest Reactiva

Nuestra aplicación Spring Boot será reactiva gracias al uso de la librería spring-boot-starter-webflux, ya que nos permite escribir y leer REST APIs reactivas. Para que podamos relacionar el concepto con algo más tangible vamos a ver un ejemplo de programación reactiva usando Spring Boot y Kotlin.

import org.springframework.http.MediaType.TEXT_EVENT_STREAM_VALUE
import reactor.core.publisher.Flux
import reactor.core.publisher.Mono
import java.time.Duration.ofSeconds

@RestController
class SongController {

    @Autowired
    private lateinit var songRepository: SongRepository

    @GetMapping(value = ["/song"], produces = [TEXT_EVENT_STREAM_VALUE])
    fun findSongs(): Flux<Song> {
        return songRepository.findAll().delayElements(ofSeconds(3))
    }

    @PostMapping(value = ["/song"], produces = [TEXT_EVENT_STREAM_VALUE])
    fun saveSong(@RequestBody song: Song): Mono<Song> {
        return songRepository.save(song)
    }

}

Para habilitar la programación reactiva en el primer endpoint, hemos usado el tipo de retorno Flux de la biblioteca Reactor de Spring. Este tipo representa un flujo de datos asincrónico que puede emitir varios valores a lo largo del tiempo. Además, hemos especificado que la API debe producir un tipo de contenido text/event-stream, lo que indica que se trata de un flujo de eventos en tiempo real. Para simular que nuestra API está tratando de consumir datos desde una fuente que puede tardar más tiempo del deseado, hemos usado delayElements para retrasar un poco el envío de datos desde nuestro endpoint.

Para que nuestra API sea totalmente reactiva, cabe destacar que hemos usado la librería R2DBC, más concretamente en su implementación para PostgreSQL, la cual nos permite operar con la base de datos de una forma reactiva trabajando con los ya comentados flujos de datos. La ventaja frente a JDBC (librería más común para persistencia en el mundillo de Spring Boot) es que los métodos que nos ofrece para operar con la base de datos no son bloqueantes y nos facilita la integración con nuestra API reactiva ya que trabaja usando los tipos de datos asíncronos como Flux o Mono. Vamos a ver la implementación del repositorio:

...
//R2DBC PostgreSQL
import org . springframework . data . r2dbc . repository . R2dbcRepository

        @Repository
        interface SongRepository : R2dbcRepository<Song, Int>

Al extender de ReactiveCrudRepository ya tenemos disponibles una gran cantidad de métodos como find, findAll, Save… (véase el ejemplo en el código del controlador más arriba).

En el segundo endpoint, hemos usado el tipo Mono, que solo contiene un valor y que usamos para retornar el elemento que guardamos en la base de datos.

Otro aspecto importante a tener en cuenta es cómo configuramos SpringBoot para que pueda conectarse a nuestra base de datos. Tenemos que especificar qué URL tiene que usar R2DBC para conectarse:

Inserta este código en tu application.yml o application.properties:

spring:
  r2dbc:
    url: r2dbc:postgresql://postgres:5432/admin
    username: admin
    password: password

Esta configuración será variable dependiendo de como tengas desplegada tu base de datos. En el ejemplo que proporciono, la base de datos se arranca en local junto con la aplicación utilizando Docker.

Consumir API Rest Reactiva

Aquí te presento un ejemplo de cómo consumir de forma reactiva esta API desde una aplicación cliente usando la biblioteca Reactor de Spring:

...
import org . springframework . web . reactive . function . client . WebClient
        import java . time . Duration

        @RestController
        class ConsumeSongController {

            @GetMapping("/consume")
            fun consumeAll() {
                val response = WebClient.builder()
                    .baseUrl(SONGS_URL)
                    .build()
                    .get()
                    .retrieve()
                    .bodyToFlux(Map::class.java)
                response.subscribe { song ->
                    println(song)
                }
            }

            @GetMapping("/consume-only-two")
            fun consumeOnlyTwo() {
                val response = WebClient.builder()
                    .baseUrl(SONGS_URL)
                    .build()
                    .get()
                    .retrieve()
                    .bodyToFlux(Map::class.java)
                response.take(2).subscribe { song ->
                    println(song)
                }
            }

            @GetMapping("/consume-for-five-seconds")
            fun consumeForFiveSeconds() {
                val response = WebClient.builder()
                    .baseUrl(SONGS_URL)
                    .build()
                    .get()
                    .retrieve()
                    .bodyToFlux(Map::class.java)
                response.take(Duration.ofSeconds(5)).subscribe { song ->
                    println(song)
                }
            }

        }

private const val SONGS_URL = "http://127.0.0.1:8080/song"

Para recibir la respuesta como un flujo de eventos, hemos usado el método bodyToFlux de la clase WebClient ( librería reactiva para consumir APIs), que nos permite almacenar los datos a medida que llegan en la variable. Finalmente, hemos suscrito un manejador al flujo de eventos, que simplemente imprime cada evento recibido por consola. Al ejecutar este código, la aplicación cliente comenzará a consumir el flujo de eventos de forma reactiva y se imprimirá un mensaje por cada evento recibido.

Como comentamos anteriormente, tenemos muchas posibilidades para evitar la sobrecarga en nuestra aplicación especificando cada cuanto tiempo queremos coger datos o cuanta cantidad, etc.

Webgrafía

Jacquot, A. (8 de julio del 2021). Programación reactiva con Spring Data R2DBC. Blog de Pictet Technologies: https://medium.com/pictet-technologies-blog/reactive-programming-with-spring-data-r2dbc-ee9f1c24848b

Paluch, M. (7 de diciembre de 2018). Programación Reactiva y Bases de Datos Relacionales. Blog de primavera: https://spring.io/blog/2018/12/07/reactive-programming-and-relational-databases

Vijay, Srj. (10 de junio del 2022). ¿Cómo implementar la programación reactiva en Spring Boot?. El desarrollador de la pila completa: https://fullstackdeveloper.guru/2022/06/10/how-to-implement-reactive-programming-in-spring-boot/