在 Spring Webflux 中执行阻塞 JDBC 调用

我使用 Spring Webflux 和 Spring data jpa 使用 PostgreSql 作为后端数据库。 我不想在进行数据库调用时阻塞主线程，例如find and 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。这种理解正确吗？

您的理解是正确的publishOn and subscribeOn (请参阅reactor项目中有关这些操作员的参考文档 https://projectreactor.io/docs/core/release/reference/#_publishon).

如果您调用阻塞库而不在特定调度程序上调度该工作，那么这些调用将阻塞少数可用线程之一（默认情况下，Netty 事件循环），并且您的应用程序将只能同时服务几个请求。

现在我不确定你这样做想达到什么目的。

首先，并行调度程序专为 CPU 密集型任务而设计 https://projectreactor.io/docs/core/release/reference/#schedulers，这意味着您将拥有很少的数量，与 CPU 核心一样多（或更多）的数量。在这种情况下，这就像将线程池大小设置为常规 Servlet 容器上的核心数。您的应用程序将无法处理大量并发请求。

即使您选择更好的替代方案（例如弹性调度程序），它仍然不如 Netty 事件循环，后者是 Spring WebFlux 中本地调度请求处理的地方。

如果您的最终目标是性能和可扩展性，那么将阻塞调用包装在响应式应用程序中可能会比常规 Servlet 容器的性能更差。

您可以改为使用 Spring MVC 并且：

• 当您处理阻塞库（如 JPA）时，使用常用的阻塞返回类型
• use Mono and Flux当您不依赖于此类库时返回类型

这不会是非阻塞的，但这仍然是异步的，您将能够并行地完成更多工作，而无需处理复杂性。