R2DBC: Reactive, Non-Blocking Relational Access

For most of its history, relational access in the JVM has meant JDBC. JDBC is a fundamentally blocking API: when you call executeQuery(), the calling thread is suspended at the operating-system level until the database answers. That model is simple to reason about, but it ties up a real platform thread for the entire duration of every query. Under a reactive stack like Spring WebFlux, where the whole point is to serve thousands of concurrent requests on a small event-loop thread pool, a single blocking JDBC call can stall the event loop and collapse your throughput. R2DBC (Reactive Relational Database Connectivity) exists to close that gap: it is a specification for talking to SQL databases without ever blocking a thread.

The core difference is in the shape of the return types. A JDBC method hands you a fully materialized ResultSet right now, because the thread waited for it. An R2DBC driver instead hands you a Publisher from the Reactive Streams specification — typically a Mono (zero or one item) or a Flux (zero to many items) when you use Project Reactor. Nothing has executed yet; the Publisher is a lazy recipe. Work only begins when something subscribes, and as rows arrive from the network they are pushed to the subscriber with backpressure, so a slow consumer never forces the driver to buffer an unbounded result set in memory. The connection itself is non-blocking: io.r2dbc.spi.Connection returns Publishers for begin, commit, and rollback rather than void.

You rarely touch the raw SPI in a Spring application. Spring Data R2DBC gives you DatabaseClient, a fluent, null-safe wrapper over a ConnectionFactory. The canonical query reads almost like English: databaseClient.sql("...").bind("name", value).fetch().all(). The sql(...) step holds the parameterized statement, bind(...) safely binds named parameters (never string concatenation, so you are protected from SQL injection), fetch() chooses how much you want back, and the terminal operator decides the cardinality — all() returns a Flux of rows, one() expects exactly one, first() takes the head, and rowsUpdated() returns the affected-row count for writes.

Here is the part that makes R2DBC pleasant in Kotlin. A bare Flux<Map<String, Object>> is awkward to consume, so Spring exposes coroutine extensions in the org.springframework.r2dbc.core package. awaitSingle(), awaitSingleOrNull(), awaitRowsUpdated(), and the Flux-to-Flow bridge asFlow() let you write straight-line suspending code that reads like blocking JDBC but never blocks a thread. Inside a suspend function the dispatcher is freed the instant you await, so the event-loop thread is handed back to serve other requests while the database does its work. You get the readability of imperative code with the scalability of the reactive runtime.

Mapping rows to objects is explicit by default. The .map { row, metadata -> ... } operator gives you a Readable row from which you pull typed columns, e.g. row.get("id", java.util.UUID::class.java). If you prefer convention over ceremony, define a data class and let Spring Data map it for you via the repository abstraction (CoroutineCrudRepository) or DatabaseClient's entity-aware fetch. Note a deliberate limitation: R2DBC is intentionally lean. It is not a full ORM — there is no lazy loading, no persistence context, and no automatic relationship graph traversal the way JPA/Hibernate offers. You compose joins and aggregates with explicit SQL, which keeps behavior predictable and the round-trips visible.

Transactions follow the same reactive discipline. Because a transaction is bound to a connection and that connection's work is asynchronous, you cannot rely on a thread-local like JDBC does. Spring solves this with reactive context propagation: annotate a suspend function with @Transactional and Spring carries the active transaction through the Reactor Context (and the coroutine context), committing when the returned publisher completes and rolling back when it errors. The mental rule is simple — the transaction lives as long as the reactive pipeline, not as long as a thread.

When should you actually reach for R2DBC? Choose it when your service is genuinely I/O-bound, already built on WebFlux or coroutines end to end, and you need high concurrency with a bounded thread pool — streaming large result sets, fan-out queries, or high-connection-count workloads. If your application is a classic blocking Spring MVC service, mixing in R2DBC buys you little and costs you complexity; plain JDBC or JPA is the better tool. R2DBC also has a smaller ecosystem and no second-level cache, so weigh those trade-offs. The decisive question is whether you can keep the entire call chain non-blocking: a single blocking call buried in a reactive pipeline negates every advantage R2DBC was designed to give you.

One last practical note: drivers are database-specific (r2dbc-postgresql, r2dbc-mariadb, r2dbc-mssql, r2dbc-h2, and so on), and you wire them through an io.r2dbc.spi.ConnectionFactory rather than a javax.sql.DataSource. Spring Boot autoconfigures this from your spring.r2dbc.url, spring.r2dbc.username, and spring.r2dbc.password properties — note the r2dbc: URL scheme, e.g. r2dbc:postgresql://localhost/app. Add the spring-boot-starter-data-r2dbc dependency, point your URL at the database, and the DatabaseClient and repository beans are ready to inject.

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
}