Spring webflux and reading from database

Question

Spring 5 introduces the reactive programming style for rest APIs with webflux. I'm fairly new to it myself and was wondering wether wrapping synchronous calls to a database into Flux or Mono makes sense preformence-wise? If yes, is this the way to do it:

@RestController public class HomeController {      private MeasurementRepository repository;      public HomeController(MeasurementRepository repository){         this.repository = repository;     }      @GetMapping(value = "/v1/measurements")     public Flux<Measurement> getMeasurements() {         return Flux.fromIterable(repository.findByFromDateGreaterThanEqual(new Date(1486980000L)));     }  } 

Is there something like an asynchronous CrudRepository? I couldn't find it.

Answer 1

One option would be to use alternative SQL clients that are fully non-blocking. Some examples include: https://github.com/mauricio/postgresql-async or https://github.com/finagle/roc. Of course, none of these drivers is officially supported by database vendors yet. Also, functionality is way much less attractive comparing to mature JDBC-based abstractions such as Hibernate or jOOQ.

The alternative idea came to me from Scala world. The idea is to dispatch blocking calls into isolated ThreadPool not to mix blocking and non-blocking calls together. This will allow us to control the overall number of threads and will let the CPU serve non-blocking tasks in the main execution context with some potential optimizations. Assuming that we have JDBC based implementation such as Spring Data JPA which is indeed blocking, we can make it’s execution asynchronous and dispatch on the dedicated thread pool.

@RestController public class HomeController {      private final MeasurementRepository repository;     private final Scheduler scheduler;      public HomeController(MeasurementRepository repository, @Qualifier("jdbcScheduler") Scheduler scheduler) {         this.repository = repository;         this.scheduler = scheduler;     }      @GetMapping(value = "/v1/measurements")     public Flux<Measurement> getMeasurements() {         return Mono.fromCallable(() -> repository.findByFromDateGreaterThanEqual(new Date(1486980000L))).publishOn(scheduler);     }  } 

Our Scheduler for JDBC should be configured by using dedicated Thread Pool with size count equal to the number of connections.

@Configuration public class SchedulerConfiguration {     private final Integer connectionPoolSize;      public SchedulerConfiguration(@Value("${spring.datasource.maximum-pool-size}") Integer connectionPoolSize) {         this.connectionPoolSize = connectionPoolSize;     }      @Bean     public Scheduler jdbcScheduler() {         return Schedulers.fromExecutor(Executors.newFixedThreadPool(connectionPoolSize));     }  } 

However, there are difficulties with this approach. The main one is transaction management. In JDBC, transactions are possible only within a single java.sql.Connection. To make several operations in one transaction, they have to share a connection. If we want to make some calculations in between them, we have to keep the connection. This is not very effective, as we keep a limited number of connections idle while doing calculations in between.

This idea of an asynchronous JDBC wrapper is not new and is already implemented in Scala library Slick 3. Finally, non-blocking JDBC may come along on the Java roadmap. As it was announced at JavaOne in September 2016, and it is possible that we will see it in Java 10.

Answer 2

Based on this blog you should rewrite your snippet in following way

@GetMapping(value = "/v1/measurements") public Flux<Measurement> getMeasurements() {     return Flux.defer(() -> Flux.fromIterable(repository.findByFromDateGreaterThanEqual(new Date(1486980000L))))            .subscribeOn(Schedulers.elastic()); }