R2DBC: acceso relacional reactivo y no bloqueante

Durante la mayor parte de su historia, el acceso relacional en la JVM ha significado JDBC. JDBC es una API fundamentalmente bloqueante: cuando llamas a executeQuery(), el hilo que llama queda suspendido a nivel del sistema operativo hasta que la base de datos responde. Ese modelo es fácil de razonar, pero retiene un hilo real de plataforma durante toda la duración de cada consulta. En un stack reactivo como Spring WebFlux, donde el objetivo es atender miles de peticiones concurrentes sobre un pequeño pool de hilos de event-loop, una sola llamada bloqueante de JDBC puede paralizar el event-loop y hundir tu rendimiento. R2DBC (Reactive Relational Database Connectivity) existe para cerrar esa brecha: es una especificación para hablar con bases de datos SQL sin bloquear nunca un hilo.

La diferencia central está en la forma de los tipos de retorno. Un método de JDBC te entrega un ResultSet completamente materializado ahora mismo, porque el hilo esperó por él. Un driver de R2DBC, en cambio, te entrega un Publisher de la especificación Reactive Streams — normalmente un Mono (cero o un elemento) o un Flux (cero a muchos elementos) cuando usas Project Reactor. Nada se ha ejecutado todavía; el Publisher es una receta perezosa. El trabajo solo empieza cuando algo se suscribe, y a medida que las filas llegan de la red se empujan al suscriptor con backpressure, de modo que un consumidor lento nunca obliga al driver a almacenar en memoria un resultado ilimitado. La propia conexión es no bloqueante: io.r2dbc.spi.Connection devuelve Publishers para begin, commit y rollback en lugar de void.

Rara vez tocas el SPI crudo en una aplicación Spring. Spring Data R2DBC te da DatabaseClient, un envoltorio fluido y null-safe sobre un ConnectionFactory. La consulta canónica se lee casi como inglés: databaseClient.sql("...").bind("name", value).fetch().all(). El paso sql(...) contiene la sentencia parametrizada, bind(...) enlaza de forma segura parámetros con nombre (nunca concatenación de cadenas, así que estás protegido contra inyección SQL), fetch() elige cuánto quieres de vuelta, y el operador terminal decide la cardinalidad — all() devuelve un Flux de filas, one() espera exactamente una, first() toma la primera, y rowsUpdated() devuelve el número de filas afectadas en escrituras.

Aquí está la parte que hace agradable a R2DBC en Kotlin. Un Flux<Map<String, Object>> pelado es incómodo de consumir, así que Spring expone extensiones de corrutinas en el paquete org.springframework.r2dbc.core. awaitSingle(), awaitSingleOrNull(), awaitRowsUpdated() y el puente Flux-a-Flow asFlow() te permiten escribir código suspendido lineal que se lee como JDBC bloqueante pero nunca bloquea un hilo. Dentro de una función suspend el dispatcher se libera en el instante en que haces await, de modo que el hilo del event-loop se devuelve para atender otras peticiones mientras la base de datos hace su trabajo. Obtienes la legibilidad del código imperativo con la escalabilidad del runtime reactivo.

El mapeo de filas a objetos es explícito por defecto. El operador .map { row, metadata -> ... } te da una fila Readable de la que extraes columnas tipadas, p. ej. row.get("id", java.util.UUID::class.java). Si prefieres convención sobre ceremonia, define una data class y deja que Spring Data la mapee por ti mediante la abstracción de repositorio (CoroutineCrudRepository) o el fetch consciente de entidades de DatabaseClient. Observa una limitación deliberada: R2DBC es intencionalmente ligero. No es un ORM completo — no hay carga perezosa, ni contexto de persistencia, ni recorrido automático del grafo de relaciones como ofrece JPA/Hibernate. Compones joins y agregaciones con SQL explícito, lo que mantiene el comportamiento predecible y los viajes a la base de datos visibles.

Las transacciones siguen la misma disciplina reactiva. Como una transacción está ligada a una conexión y el trabajo de esa conexión es asíncrono, no puedes apoyarte en un thread-local como hace JDBC. Spring lo resuelve con propagación reactiva de contexto: anota una función suspend con @Transactional y Spring transporta la transacción activa a través del Reactor Context (y el contexto de la corrutina), confirmando cuando el publisher devuelto completa y revirtiendo cuando falla. La regla mental es simple — la transacción vive tanto como el pipeline reactivo, no tanto como un hilo.

¿Cuándo deberías recurrir realmente a R2DBC? Elígelo cuando tu servicio sea genuinamente intensivo en E/S, ya esté construido sobre WebFlux o corrutinas de extremo a extremo, y necesites alta concurrencia con un pool de hilos acotado — streaming de grandes resultados, consultas en abanico, o cargas con muchas conexiones. Si tu aplicación es un servicio clásico bloqueante de Spring MVC, mezclar R2DBC te aporta poco y te cuesta complejidad; JDBC o JPA simples son la mejor herramienta. R2DBC también tiene un ecosistema más pequeño y sin caché de segundo nivel, así que sopesa esos compromisos. La pregunta decisiva es si puedes mantener toda la cadena de llamadas no bloqueante: una sola llamada bloqueante enterrada en un pipeline reactivo anula toda ventaja para la que R2DBC fue diseñado.

Una última nota práctica: los drivers son específicos de cada base de datos (r2dbc-postgresql, r2dbc-mariadb, r2dbc-mssql, r2dbc-h2, etc.), y los conectas mediante un io.r2dbc.spi.ConnectionFactory en lugar de un javax.sql.DataSource. Spring Boot lo autoconfigura desde tus propiedades spring.r2dbc.url, spring.r2dbc.username y spring.r2dbc.password — fíjate en el esquema de URL r2dbc:, p. ej. r2dbc:postgresql://localhost/app. Añade la dependencia spring-boot-starter-data-r2dbc, apunta tu URL a la base de datos, y los beans DatabaseClient y de repositorio quedan listos para inyectar.

import kotlinx.coroutines.flow.Flow
import org.springframework.r2dbc.core.DatabaseClient
import org.springframework.r2dbc.core.awaitRowsUpdated
import org.springframework.r2dbc.core.awaitSingleOrNull
import org.springframework.r2dbc.core.flow
import org.springframework.stereotype.Repository
import java.util.UUID

data class Book(val id: UUID, val title: String, val authorId: UUID)

@Repository
class BookRepository(private val client: DatabaseClient) {

    // Flux<row> bridged to a Kotlin Flow; rows are pushed with backpressure.
    fun findByAuthor(authorId: UUID): Flow<Book> =
        client.sql("SELECT id, title, author_id FROM book WHERE author_id = :author")
            .bind("author", authorId)
            .map { row, _ ->
                Book(
                    id = row.get("id", UUID::class.java)!!,
                    title = row.get("title", String::class.java)!!,
                    authorId = row.get("author_id", UUID::class.java)!!,
                )
            }
            .flow()

    // Mono<row> awaited as a suspend call; null when nothing matches.
    suspend fun findById(id: UUID): Book? =
        client.sql("SELECT id, title, author_id FROM book WHERE id = :id")
            .bind("id", id)
            .map { row, _ ->
                Book(
                    id = row.get("id", UUID::class.java)!!,
                    title = row.get("title", String::class.java)!!,
                    authorId = row.get("author_id", UUID::class.java)!!,
                )
            }
            .awaitSingleOrNull()

    // Writes: fetch().awaitRowsUpdated() suspends and returns the affected-row count.
    suspend fun insert(book: Book): Long =
        client.sql("INSERT INTO book (id, title, author_id) VALUES (:id, :title, :author)")
            .bind("id", book.id)
            .bind("title", book.title)
            .bind("author", book.authorId)
            .fetch()
            .awaitRowsUpdated()
}

import kotlinx.coroutines.flow.Flow
import org.springframework.data.annotation.Id
import org.springframework.data.r2dbc.repository.Query
import org.springframework.data.relational.core.mapping.Table
import org.springframework.data.repository.kotlin.CoroutineCrudRepository
import org.springframework.stereotype.Service
import org.springframework.transaction.annotation.Transactional
import java.util.UUID

@Table("book")
data class BookEntity(
    @Id val id: UUID? = null,
    val title: String,
    val authorId: UUID,
)

interface BookCrudRepository : CoroutineCrudRepository<BookEntity, UUID> {
    fun findAllByAuthorId(authorId: UUID): Flow<BookEntity>

    @Query("SELECT * FROM book WHERE title ILIKE :pattern")
    fun search(pattern: String): Flow<BookEntity>
}

@Service
class CatalogService(private val books: BookCrudRepository) {
    // The transaction lives as long as the suspending call's pipeline.
    @Transactional
    suspend fun rename(id: UUID, newTitle: String): BookEntity? {
        val existing = books.findById(id) ?: return null
        return books.save(existing.copy(title = newTitle))
    }
}

import kotlinx.coroutines.*
import kotlin.system.measureTimeMillis

@OptIn(DelicateCoroutinesApi::class)
suspend fun main() {
    val queries = 8
    val latencyMs = 200L // pretend each DB round-trip takes 200ms

    // Blocking style: a pool of 2 "threads" forces serialized waves.
    val pool = newFixedThreadPoolContext(2, "db-pool")
    val blocking = measureTimeMillis {
        coroutineScope {
            repeat(queries) {
                launch(pool) { Thread.sleep(latencyMs) } // holds the thread
            }
        }
    }
    pool.close()

    // Non-blocking style: delay() suspends instead of parking a thread.
    val nonBlocking = measureTimeMillis {
        coroutineScope {
            repeat(queries) {
                launch { delay(latencyMs) } // frees the dispatcher while waiting
            }
        }
    }

    println("blocking (2 threads):  " + blocking + " ms")     // ~4 waves of 200ms
    println("non-blocking:          " + nonBlocking + " ms")  // ~200ms, all overlap
}