Выполнить блокировку вызова JDBC в Spring Webflux - PullRequest
Новая
       35

Выполнить блокировку вызова JDBC в Spring Webflux

0 голосов
/ 25 февраля 2019

Я использую Spring Webflux с jpa-файлами данных Spring, используя PostgreSql в качестве базы данных.Я не хочу блокировать основной поток при выполнении вызовов БД, таких как find и save.Чтобы добиться того же, у меня есть главный планировщик в классе Controller и классы обслуживания jdbcScheduler.

Я определил их следующим образом: 

@Configuration
@EnableJpaAuditing
public class CommonConfig {

    @Value("${spring.datasource.hikari.maximum-pool-size}")
    int connectionPoolSize;

    @Bean
    public Scheduler scheduler() {
        return Schedulers.parallel();
    }

    @Bean
    public Scheduler jdbcScheduler() {
        return Schedulers.fromExecutor(Executors.newFixedThreadPool(connectionPoolSize));
    }

    @Bean
    public TransactionTemplate transactionTemplate(PlatformTransactionManager transactionManager) {
        return new TransactionTemplate(transactionManager);
    }
}

Теперь, пока я делаювызов get / save в моем слое обслуживания я делаю: 

    @Override
    public Mono<Config> getConfigByKey(String key) {
        return Mono.defer(
            () -> Mono.justOrEmpty(configRepository.findByKey(key)))
            .subscribeOn(jdbcScheduler)
            .publishOn(scheduler);
    }

    @Override
    public Flux<Config> getAllConfigsAfterAppVersion(int appVersion) {
        return Flux
            .fromIterable(configRepository.findAllByMinAppVersionIsGreaterThanEqual(appVersion))
            .subscribeOn(jdbcScheduler)
            .publishOn(scheduler);
    }

    @Override
    public Flux<Config> addConfigs(List<Config> configList) {
        return Flux.fromIterable(configRepository.saveAll(configList))
            .subscribeOn(jdbcScheduler)
            .publishOn(scheduler);
    }

А в контроллере я делаю: 

    @PostMapping
    @ResponseStatus(HttpStatus.CREATED)
    Mono<ResponseDto<List<Config>>> addConfigs(@Valid @RequestBody List<Config> configs) {
        return configService.addConfigs(configs).collectList()
            .map(configList -> new ResponseDto<>(HttpStatus.CREATED.value(), configList, null))
            .subscribeOn(scheduler);
    }

Это правильно?и / или есть лучший способ сделать это?

Что я понимаю под: 

.subscribeOn(jdbcScheduler)
.publishOn(scheduler);

, так это то, что задача будет выполняться в потоках jdbcScheduler, а последующий результат будет опубликован вмоя основная параллель scheduler.Это понимание правильно?

1 Ответ

0 голосов
/ 25 февраля 2019

Ваше понимание верно в отношении publishOn и subscribeOn ( см. Справочную документацию в проекте реактора об этих операторах ).

Если вы вызываете блокирующие библиотеки без планирования,При работе с определенным планировщиком эти вызовы блокируют один из немногих доступных потоков (по умолчанию - цикл событий Netty), и ваше приложение сможет одновременно обслуживать только несколько запросов.

Теперь я неубедитесь, что вы пытаетесь достичь, делая это.

Во-первых, параллельный планировщик предназначен для задач, связанных с процессором , то есть их будет мало, столько же (илинемного больше) в качестве процессорных ядер.В этом случае это все равно, что установить размер пула потоков равным количеству ядер в обычном контейнере сервлетов.Ваше приложение не сможет обрабатывать большое количество одновременных запросов.

Даже если вы выберете лучшую альтернативу (например, эластичный планировщик), оно все равно будет не так хорошо, как цикл событий Netty, которыйименно здесь обработка запросов запланирована изначально в Spring WebFlux.

Если ваша конечная цель - производительность и масштабируемость, то блокирование вызовов в реагирующем приложении может работать хуже, чем в обычном контейнере сервлетов.

Вместо этого вы могли бы использовать Spring MVC и:

  • использовать обычные типы блокирующих возвратов, когда вы имеете дело с библиотекой блокировок, например JPA
  • , использовать Mono и Flux возвращаемые типыкогда вы не привязаны к таким библиотекам

Это не будет неблокирующим, но все равно будет асинхронным, и вы сможете параллельно выполнять больше работы, не сталкиваясь со сложностью..

...