Реактивное программирование

R2DBC vs JDBC: реактивные базы данных

Исторический контекст: что такое JDBC и почему он доминировал десятилетиями

JDBC — это стандартный API Java для доступа к реляционным базам данных, появившийся ещё в JDK 1.1 (1997 год).

Он позволяет выполнять SQL-запросы, управлять соединениями и обрабатывать результаты через унифицированный интерфейс, независимо от конкретной БД (PostgreSQL, MySQL, Oracle и т.д.).

Ключевые компоненты JDBC:
DriverManager или DataSource: Для получения соединения (Connection).
Statement/PreparedStatement: Для выполнения SQL (executeQuery, executeUpdate).
ResultSet: Для чтения результатов (next(), getString() и т.д.).
Transaction management: commit(), rollback().

Пример простого JDBC-кода:

import java.sql.*;
public class JdbcExample {
  public static void main(String[] args) {
     try (Connection conn = DriverManager.getConnection("jdbc:postgresql://localhost:5432/db", "user", "pass");
     PreparedStatement stmt = conn.prepareStatement("SELECT * FROM users WHERE id = ?")) {
     stmt.setLong(1, 1L);
     ResultSet rs = stmt.executeQuery();

     while (rs.next()) {
          System.out.println("User: " + rs.getString("name"));
      }
   } catch (SQLException e) {
        e.printStackTrace();
    }
  }
}

Это синхронно и блокирующе: executeQuery() "виснет" до ответа от БД, блокируя текущий поток.

В традиционных приложениях (как Spring MVC) это работало: каждый запрос — отдельный поток из пула (например, Tomcat с 200 потоками), и если БД отвечает быстро, проблем нет. Но под высокой нагрузкой или с медленными запросами (сетевые задержки, сложные джойны) пул исчерпывается: потоки "спят" в ожидании IO, CPU простаивает, а новые запросы ждут в очереди, вызывая таймауты и отказы. Это классическая проблема асинхронщины: JDBC не предназначен для non-blocking IO, он полагается на blocking calls операционной системы.

В реактивных приложениях (WebFlux) использование JDBC — антипаттерн: если контроллер возвращает Mono, но внутри — blocking JDBC, весь выигрыш теряется. Поток из event-loop (Netty) блокируется, снижая throughput (пропускную способность). Вот почему нужен новый подход.


Проблемы JDBC в реактивном контексте: почему старый стандарт не справляется

Давайте разберём проблемы JDBC подробно, чтобы понять мотивацию R2DBC:
Блокирующая природа: Все операции (connect, query, fetch) — синхронны. В асинхронном коде это требует обёрток вроде CompletableFuture или offload на отдельный пул (Schedulers.boundedElastic()), но это хак: теряется истинная реактивность, и под нагрузкой пулы переполняются.

Отсутствие backpressure: ResultSet — pull-модель (next() получает данные), но без контроля темпа. Если результат огромный (миллионы строк), буфер переполняется, рискуя OOM (OutOfMemoryError). В реактивном мире (push с backpressure) это несовместимо.

Управление соединениями: JDBC полагается на пулы (HikariCP), но они ориентированы на blocking: соединение "занято" весь запрос. В реактиве нужно multiplexing — одно соединение для многих операций асинхронно.

Транзакции: @Transactional в Spring работает, но в реактиве требует специальной поддержки (reactive transactions), иначе — блокировки.

Масштабируемость: Под 10k+ RPS (requests per second) с БД-запросами JDBC требует огромных пулов потоков (тысячи), что жрёт память (каждый поток ~1MB стека) и контекст-свичинг.

Интеграция с Reactor: Нет native Publisher — результаты не "текут" как Flux, требуя ручной конвертации, что добавляет boilerplate и риски.

В итоге, JDBC — отличный для legacy или низконагруженных приложений, но в микросервисах с WebFlux он "ломает" реактивный стек, возвращая к болям callback-ада и ожиданий.


Введение в R2DBC: реактивный стандарт для реляционных БД

R2DBC — это спецификация (с 2019 года, под эгидой Spring и Pivotal), определяющая API для доступа к реляционным БД в реактивном стиле. Это не замена JDBC, а параллельный стандарт, ориентированный на non-blocking IO.

#Java #middle #Reactor #WebFlux #Mono #Flux #R2DBC

Ключевые идеи:
Publisher-based API: Все операции возвращают Publisher (Mono/Flux из Reactive Streams): Connection как Mono<Connection>, Statement.execute() как Flux<Row>.

Non-blocking от начала до конца: Использует асинхронные драйверы (для PostgreSQL, MySQL и т.д.), где соединения мультиплексируются — одно для многих запросов.

Backpressure встроено: Результаты (Flux<Row>) уважают request(n): если подписчик не готов, БД не шлёт данные, избегая перегрузки.

Транзакции реактивные: Поддержка @Transactional с Mono/Flux.

Интеграция с экосистемой: Spring Data R2DBC — аналог Spring Data JPA, с репозиториями, @Query и CRUD.

Драйверы: r2dbc-postgresql, r2dbc-mysql и т.д. — реализуют спецификацию, используя неблокирующие сокеты (Netty или аналог).

Пример базового R2DBC-кода (без Spring):

import io.r2dbc.spi.ConnectionFactories;
import io.r2dbc.spi.ConnectionFactory;
import reactor.core.publisher.Flux;

public void createConnectionFactory () {

  ConnectionFactory factory = ConnectionFactories.get("r2dbc:postgresql://localhost:5432/db?username=user&password=pass");

  Flux<String> namesFlux = Flux.usingWhen(

     factory.create(), // Асинхронно создать соединение

     conn -> conn.createStatement("SELECT name FROM users").execute().flatMap(result -> result.map((row, metadata) -> row.get("name", String.class))),
     conn -> conn.close() // Асинхронно закрыть
  );

  namesFlux.subscribe(System.out::println); // Строки приходят асинхронно
}

Здесь usingWhen — реактивный try-with-resources: создаёт соединение асинхронно, выполняет запрос как Flux<Result>, map извлекает данные. Нет блокировок: если БД медленная, поток свободен.


Spring Data R2DBC: упрощение с репозиториями и аннотациями

Spring Data R2DBC — модуль, который абстрагирует R2DBC, как Spring Data JPA для JDBC.

Добавьте зависимость:

<dependency>
  <groupId>org.springframework.boot</groupId>
  <artifactId>spring-boot-starter-data-r2dbc</artifactId>
</dependency>
<dependency>
  <groupId>io.r2dbc</groupId>
  <artifactId>r2dbc-postgresql</artifactId> <!-- Для PostgreSQL -->
</dependency>

Настройте в application.properties:

spring.r2dbc.url=r2dbc:postgresql://localhost:5432/db
spring.r2dbc.username=user
spring.r2dbc.password=pass

Репозитории:

ReactiveRepository extends ReactiveCrudRepository<Entity, ID>.

Сущность:

@Entity
public class User {
  @Id
  private Long id;
  private String name;
  // Getters/setters
}

public interface UserRepository extends ReactiveCrudRepository<User, Long> {
  @Query("SELECT * FROM users WHERE name LIKE :name")
  Flux<User> findByNameLike(String name);
}

В сервисе/контроллере:

@Service
public class UserService {
  private final UserRepository repo;

  public UserService(UserRepository repo) {
     this.repo = repo;
  }

  public Flux<User> findAll() {
    return repo.findAll(); // Flux асинхронно
  }

  public Mono<User> save(User user) {
     return repo.save(user);
  }
}

В контроллере:

@GetMapping("/users")
public Flux<User> getAllUsers() {
     return userService.findAll();
}

Это декларативно: repo.findAll() — Flux, который "течёт" из БД без блокировок. Транзакции: @Transactional на методе — reactive, rollback асинхронно.

Расширенный пример: пагинация с ReactiveSortingRepository и Pageable.

public interface UserRepository extends ReactiveSortingRepository<User, Long> {}
Flux<User> paged = repo.findAll(Sort.by("name").ascending()).skip(10).take(20); // Простая пагинация
Для complex: используйте @Query с параметрами, или Criteria API.

Практические советы и подводные камни

Выбор БД: PostgreSQL — лучший для R2DBC (полная поддержка async).
Тестирование: Embedded H2 с r2dbc-h2, ReactiveTest для StepVerifier.
Камень: Нет full ORM (как JPA entities с relations) — используйте ручные joins или Spring Data Projections.
Камень: Транзакции не поддерживают propagation в nested методах fully — будьте осторожны.
Совет: Для hybrid (JDBC + R2DBC) — используйте разные DataSource, но избегайте в одном приложении.
Совет: Мониторьте с Micrometer: метрики на запросы, соединения.


#Java #middle #Reactor #WebFlux #Mono #Flux #R2DBC