Какие плюсы и минусы Project Reactor при работе с БД?

Project Reactor при работе с БД: плюсы и минусы

Project Reactor — это реактивная библиотека, основанная на стандарте Reactive Streams, которая позволяет писать асинхронный, non-blocking код. При работе с базами данных это обещает значительные преимущества, но существуют и существенные ограничения.

Архитектура Reactor для БД

// Реактивная цепочка операций с БД
@RestController
public class UserController {
    private final R2dbcRepository<User, Long> userRepository;
    
    // Возвращаем Mono/Flux вместо объектов
    @GetMapping("/users/{id}")
    public Mono<User> getUser(@PathVariable Long id) {
        return userRepository.findById(id)
            .switchIfEmpty(Mono.error(new UserNotFoundException()));
    }
    
    @GetMapping("/users")
    public Flux<User> getAllUsers() {
        return userRepository.findAll();
    }
    
    @GetMapping("/users/active")
    public Flux<User> getActiveUsers() {
        return userRepository.findAll()
            .filter(u -> u.isActive())
            .delayElement(Duration.ofMillis(10));  // Simulate processing
    }
}

Плюсы Project Reactor при работе с БД

1. Неблокирующее выполнение (Non-blocking)

// С обычным JDBC (blocking):
public List<User> getAllUsers() {
    // Поток ЖДЁТ пока БД ответит
    return userRepository.findAll();  // Thread blocked!
}

// С Project Reactor (non-blocking):
public Flux<User> getAllUsers() {
    // Поток НЕ ждёт, может обработать другие запросы
    return userRepository.findAll();  // Thread free!
}

Преимущества:

• Один поток может обработать множество запросов
• Масштабируемость без создания новых потоков
• Лучше использование ресурсов
• Высокая throughput

Бенчмарк:

Традиционный подход (JDBC):
- 100 threads, 10,000 requests
- Memory: 50MB (100 thread stacks)
- CPU: высокое переключение контекста

Reactor подход (R2DBC):
- 10 threads (event loop), 10,000 requests
- Memory: 5MB (10 thread stacks + event queue)
- CPU: минимальное переключение контекста

2. Обработка потоков данных (Backpressure)

// Проблема: продюсер генерирует данные быстрее чем консьюмер может обработать
public Flux<User> processLargeDataset() {
    return userRepository.findAll()  // Может вернуть миллионы строк
        .buffer(1000)  // Обработать порциями по 1000
        .flatMap(batch -> processBatch(batch));  // Не перегружаем систему
}

// Reactor автоматически:
// 1. Запрашивает только то что может обработать (demand)
// 2. Паузит источник если буфер полный
// 3. Возобновляет когда снова есть место

Преимущества:

• Автоматический контроль потока
• Нет OutOfMemoryError при большых dataset
• Предсказуемое использование памяти
• Graceful degradation под нагрузкой

3. Декларативная цепочка операций

// Реактивные операции очень читаемы
public Flux<UserSummary> getActiveUsersSummary() {
    return userRepository.findAll()
        .filter(u -> u.isActive())
        .filterWhen(u -> verifyUser(u))  // Async verification
        .map(u -> new UserSummary(u))
        .sort(Comparator.comparing(UserSummary::getName))
        .distinct(UserSummary::getId)
        .take(100)
        .timeout(Duration.ofSeconds(5))
        .doOnNext(summary -> logger.info("Processing: {}", summary))
        .onErrorResume(ex -> handleError(ex));
}

// Чтение слева направо - очень интуитивно

Преимущества:

• Функциональный стиль
• Чёткая цепочка трансформаций
• Легко добавлять операции
• Меньше boilerplate кода

4. Интеграция с WebFlux

// Spring WebFlux автоматически конвертирует Mono/Flux в JSON
@RestController
public class UserController {
    @GetMapping("/users/{id}")
    public Mono<ResponseEntity<User>> getUser(@PathVariable Long id) {
        return userRepository.findById(id)
            .map(ResponseEntity::ok)
            .defaultIfEmpty(ResponseEntity.notFound().build());
    }
    
    @GetMapping("/users")
    public Flux<User> getAllUsers() {
        return userRepository.findAll();
        // WebFlux автоматически:
        // 1. Передаёт в response как JSON stream
        // 2. Управляет backpressure
        // 3. Закрывает connection когда готово
    }
    
    @PostMapping("/users/bulk")
    public Mono<Void> bulkInsert(@RequestBody Flux<User> users) {
        return users
            .flatMap(userRepository::save)
            .then();  // Return when all complete
    }
}

Преимущества:

• Асинхронность от API до БД
• Единая реактивная система
• Нет blocking operation
• Лучшая utilisateur experience (streaming responses)

5. Retry и fallback стратегии

// Reactor делает обработку errors простой и мощной
public Mono<User> getUserWithRetry(Long id) {
    return userRepository.findById(id)
        .retryWhen(errors ->
            errors.zipWith(Flux.range(1, 3))
                .flatMap(t -> {
                    int retry = t.getT2();
                    Exception error = t.getT1();
                    if (retry < 3) {
                        long waitMs = (long) Math.pow(2, retry) * 1000;  // Exponential backoff
                        logger.info("Retry {} after {} ms", retry, waitMs);
                        return Mono.delay(Duration.ofMillis(waitMs));
                    }
                    return Mono.error(error);
                })
        )
        .onErrorResume(ex -> {
            logger.error("Failed to get user", ex);
            return Mono.just(User.SYSTEM_USER);  // Fallback
        });
}

// Это было бы адом с традиционным кодом!

Минусы Project Reactor при работе с БД

1. R2DBC драйверы очень ограничены

// Проблема: не все БД поддерживают R2DBC
Доступные драйверы:
✓ PostgreSQL (r2dbc-postgresql)
✓ MySQL (r2dbc-mysql)  
✓ H2 (r2dbc-h2)
✓ MSSQL (r2dbc-mssql)

✗ Oracle (нет!
✗ DB2 (нет!)
✗ SQLite (нет!)
✗ MongoDB (отдельно, но не R2DBC)
✗ Cassandra (отдельно)

// Если нужна Oracle - не можешь использовать Reactor

Проблемы:

• Ограниченная поддержка БД
• Proprietary драйверы иногда неполные
• Миграция с обычной БД может быть сложной

2. Кривая обучения и сложность debug

// Сложный для понимания код
public Flux<User> complexQuery() {
    return userRepository.findAll()
        .flatMap(user -> {
            return addressRepository.findByUserId(user.getId())
                .collectList()
                .map(addresses -> {
                    user.setAddresses(addresses);
                    return user;
                });
        })
        .filter(user -> user.getAddresses().size() > 0)
        .flatMapMany(user ->
            userRepository.findFriends(user.getId())
                .collectList()
                .flatMapMany(friends -> Flux.just(
                    new UserWithFriends(user, friends)
                ))
        );
}

// Debug сложный - стек вызовов нечитаемый
// Error messages не всегда понятные
// Нужно понимать Event Loop, Threads, Backpressure

Проблемы:

• Крутая кривая обучения
• Трудный debug
• Нечитаемые stack traces
• Нужна хорошая документация

3. Меньше контроля и оптимизаций

// С JDBC можешь точно контролировать
try (Connection conn = dataSource.getConnection()) {
    try (Statement stmt = conn.createStatement()) {
        stmt.setFetchSize(1000);  // Batch size
        stmt.setQueryTimeout(30);  // Timeout
        // Максимальный контроль
    }
}

// С R2DBC меньше контроля
var connection = connectionPool.create();  // Что-то происходит под капотом
// Нельзя точно управлять batch size
// Нельзя точно управлять timeouts
// Connection pool управляется автоматически

Проблемы:

• Меньше fine-tuning возможностей
• Hard to optimize для специфичных сценариев
• Может быть неэффективно для некоторых queries

4. Проблемы с транзакциями

// Транзакции с Reactor - это боль
public Mono<Void> transferMoney(Long fromId, Long toId, BigDecimal amount) {
    return connectionFactory.create()
        .flatMapMany(connection -> {
            String startTx = "START TRANSACTION";
            return connection.createStatement(startTx).execute()
                .then(() -> updateBalance(connection, fromId, amount.negate()))
                .then(() -> updateBalance(connection, toId, amount))
                .then(() -> connection.createStatement("COMMIT").execute())
                .onErrorResume(ex -> 
                    connection.createStatement("ROLLBACK")
                        .execute()
                        .then(Mono.error(ex))
                );
        })
        .then();
}

// Это ужасно! С обычным Spring Data:
транзакции - просто @Transactional!

Проблемы:

• Транзакции очень сложные
• Нет простого @Transactional
• Нужно управлять connection вручную
• Легко ошибиться

5. Нет поддержки JPA

// JPA не поддерживает асинхронность
// Нельзя делать
@Entity
public class User {
    @ManyToOne
    private Department department;  // Lazy loading?!
}

// С Reactor нужно использовать:
// 1. Spring Data R2DBC (очень базовый ORM)
// 2. Низкоуровневый SQL
// 3. Третьесторонние библиотеки (jOOQ, Querydsl)

// Результат: теряем удобство JPA

Проблемы:

• JPA не работает с Reactor
• Spring Data R2DBC очень базовый
• Много вручного SQL
• Сложнее сложные queries

6. Performance не всегда лучше

// Сценарий 1: Много простых queries
// JDBC (blocking) часто быстрее!
// Потому что overhead от Reactor и Event Loop>

// Сценарий 2: Один дорогой query
// JDBC лучше - используешь все CPU cores
// Reactor использует event loop threads

// Сценарий 3: I/O bound операции
// Reactor намного лучше
// Может обработать больше параллельных requests

// Истина: Performance зависит от use case

Проблемы:

• Не universally faster
• Overhead может быть significant
• Нужно benchmark для своего сценария
• Может быть медленнее!

7. Connection Pool проблемы

// Типичная ошибка: connection pool exhaustion
@Service
public class UserService {
    @Autowired
    private UserRepository userRepository;
    
    public Flux<User> getAllUsersWithOrders() {
        return userRepository.findAll()
            .flatMap(user -> {
                // Каждый user -> новый query в БД
                // Если 10,000 users -> 10,000 connections!
                return orderRepository.findByUserId(user.getId());
            });  // N+1 problem!
    }
}

// Результат:
// - Connection pool exhausted
// - Hanging connections
// - Memory leaks
// - Все висит

// Должно быть:
return userRepository.findAllWithOrders();  // One query!

Проблемы:

• N+1 queries легко создавать
• Connection pool ограничен
• Deadlock возможен
• Нужно быть осторожным

Сравнительная таблица

Когда использовать Project Reactor с БД

ИСПОЛЬЗУЙ когда:

• ✓ Нужна высокая throughput (тысячи одновременных requests)
• ✓ БД поддерживает R2DBC (PostgreSQL, MySQL)
• ✓ I/O bound операции
• ✓ Streaming responses (большой dataset)
• ✓ Team знает Reactive Streams

НЕ ИСПОЛЬЗУЙ когда:

• ✗ Используешь Oracle, DB2, SQLite
• ✗ Нужны сложные JPA queries
• ✗ Team не знает Reactor
• ✗ Приложение простое (несколько thousand requests/sec)
• ✗ Нужны простые транзакции

Best Practices

// 1. Избегай N+1 queries
// ПЛОХО
user.flatMap(u -> orderRepository.findByUserId(u.getId()))

// ХОРОШО
userRepository.findAllWithOrders()

// 2. Используй distinct() для дедупликации
Flux.just(1,1,2,2,3).distinct()  // [1,2,3]

// 3. Manage backpressure явно
.buffer(1000)  // Process in batches

// 4. Добавляй timeouts
.timeout(Duration.ofSeconds(5))

// 5. Используй retry с exponential backoff
.retryWhen(errors -> errors.zipWith(Flux.range(1, 3))
    .delayElement(Duration.ofSeconds(1)))

// 6. Логируй операции для debug
.doOnNext(item -> logger.debug("Item: {}", item))
.doOnError(ex -> logger.error("Error", ex))

// 7. Проверяй pool configuration
spring.r2dbc.pool.max-acquire-time=3000
spring.r2dbc.pool.max-idle-time=900000
spring.r2dbc.pool.max-lifetime=1800000

Итого

Project Reactor с БД:

Плюсы:

• Отличная scalability для I/O bound операций
• Non-blocking архитектура
• Встроенное backpressure handling
• Функциональный, читаемый код
• Интеграция с WebFlux

Минусы:

• Ограниченная поддержка БД (только R2DBC)
• Сложность транзакций
• Нет JPA
• Крутая кривая обучения
• Performance не universally лучше

Рекомендация: Reactor - отличный выбор для high-throughput, I/O-bound систем с поддерживаемой БД. Но не используй если:

• Legacy БД без R2DBC
• Team неопытная
• Не нужна высокая scalability

Для большинства CRUD приложений обычный Spring Data JPA проще, быстрее в разработке и вполне достаточно.