Как обратиться к реляционной базе данных, в которой нет реактивного драйвера

# Доступ к БД без реактивного драйвера в Spring WebFlux

Если вы используете Spring WebFlux или Project Reactor, но БД поддерживает только блокирующие драйверы (JDBC), есть несколько подходов.

1. Schedulers.boundedElastic() (рекомендуется)

Самый простой и эффективный способ — использовать виртуальные потоки с boundedElastic() scheduler:

@Repository
public class UserRepository {
    private final JdbcTemplate jdbcTemplate;

    public UserRepository(JdbcTemplate jdbcTemplate) {
        this.jdbcTemplate = jdbcTemplate;
    }

    // Возвращаем Mono, но операция выполняется в отдельном потоке
    public Mono<User> findById(Long id) {
        return Mono.fromCallable(() -> 
            jdbcTemplate.queryForObject(
                "SELECT * FROM users WHERE id = ?",
                new Object[]{id},
                new UserRowMapper()
            )
        ).subscribeOn(Schedulers.boundedElastic());
    }

    public Flux<User> findAll() {
        return Flux.fromIterable(
            jdbcTemplate.query(
                "SELECT * FROM users",
                new UserRowMapper()
            )
        ).subscribeOn(Schedulers.boundedElastic());
    }

    public Mono<Void> save(User user) {
        return Mono.fromRunnable(() -> 
            jdbcTemplate.update(
                "INSERT INTO users (name, email) VALUES (?, ?)",
                user.getName(), user.getEmail()
            )
        ).subscribeOn(Schedulers.boundedElastic())
        .then();
    }
}

2. Создание Custom Scheduler

Для лучшего контроля над потоками:

@Configuration
public class DatabaseSchedulerConfig {
    
    @Bean
    public Scheduler databaseScheduler() {
        return Schedulers.newBoundedElastic(
            10,                  // threadCap: макс. 10 потоков
            100,                 // queuedTaskCap: макс. 100 задач в очереди
            "db-thread",         // threadPoolName
            60                   // ttlSeconds: время жизни потока
        );
    }
}

Использование:

@Service
public class UserService {
    private final UserRepository userRepository;
    private final Scheduler databaseScheduler;

    public UserService(UserRepository userRepository, 
                       @Qualifier("databaseScheduler") Scheduler databaseScheduler) {
        this.userRepository = userRepository;
        this.databaseScheduler = databaseScheduler;
    }

    public Flux<User> getAllUsers() {
        return userRepository.findAll()
            .subscribeOn(databaseScheduler);
    }
}

3. Reaktive Streams Bridge (r2dbc-pool)

Если хотите настоящую реактивность, используйте R2DBC:

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-r2dbc</artifactId>
</dependency>
<dependency>
    <groupId>org.postgresql</groupId>
    <artifactId>r2dbc-postgresql</artifactId>
</dependency>

Конфигурация:

@Configuration
public class R2dbcConfiguration {
    
    @Bean
    public ConnectionFactory connectionFactory() {
        return PostgresqlConnectionFactory.builder()
            .host("localhost")
            .port(5432)
            .database("mydb")
            .username("user")
            .password("password")
            .build();
    }
}

Repository:

@Repository
public interface UserR2dbcRepository extends ReactiveCrudRepository<User, Long> {
    Mono<User> findByEmail(String email);
    Flux<User> findByStatusOrderByCreatedAtDesc(String status);
}

4. Thread Pool Executor с completableFuture

Для Spring MVC, который нужно интегрировать с WebFlux:

@Service
public class UserService {
    private final JdbcTemplate jdbcTemplate;
    private final ExecutorService executorService;

    public UserService(JdbcTemplate jdbcTemplate) {
        this.jdbcTemplate = jdbcTemplate;
        this.executorService = Executors.newFixedThreadPool(10);
    }

    public Mono<User> findById(Long id) {
        return Mono.fromFuture(
            CompletableFuture.supplyAsync(() -> 
                jdbcTemplate.queryForObject(
                    "SELECT * FROM users WHERE id = ?",
                    new Object[]{id},
                    new UserRowMapper()
                ),
                executorService
            )
        );
    }
}

5. Практический пример: REST контроллер

@RestController
@RequestMapping("/api/users")
public class UserController {
    private final UserService userService;

    public UserController(UserService userService) {
        this.userService = userService;
    }

    // GET /api/users/123
    @GetMapping("/{id}")
    public Mono<ResponseEntity<User>> getUserById(@PathVariable Long id) {
        return userService.findById(id)
            .map(ResponseEntity::ok)
            .onErrorReturn(ResponseEntity.notFound().build());
    }

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

    // POST /api/users
    @PostMapping
    public Mono<ResponseEntity<User>> createUser(@RequestBody User user) {
        return userService.save(user)
            .map(saved -> ResponseEntity.status(201).body(saved));
    }
}

6. Важные моменты

Проблемы с boundedElastic()

// ❌ НЕПРАВИЛЬНО: создаёшь новый scheduler каждый раз
public Mono<User> find(Long id) {
    return Mono.fromCallable(() -> query(id))
        .subscribeOn(Schedulers.newBoundedElastic(10, 100, "db"));
}

// ✅ ПРАВИЛЬНО: переиспользуй scheduler
@Bean
public Scheduler databaseScheduler() {
    return Schedulers.boundedElastic();
}

Обработка исключений

public Mono<User> findById(Long id) {
    return Mono.fromCallable(() -> query(id))
        .subscribeOn(Schedulers.boundedElastic())
        .onErrorMap(SQLException.class, e -> 
            new RuntimeException("Database error: " + e.getMessage(), e)
        );
}

Сравнение подходов

Вывод: используйте boundedElastic() для простых решений, но в production переходите на R2DBC или custom scheduler с правильной конфигурацией потоков.