java - 在项目 react 器中包装阻塞 I/O

Question

我有一个 spring-webflux API，它在服务层需要从使用 JDBC 的现有存储库中读取。

在阅读了有关该主题的一些内容后，我希望将阻塞数据库调用的执行与其余非阻塞异步代码分开。

我定义了一个专用的jdbcScheduler:

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

以及使用它的 AsyncWrapper 实用程序:

@Component
public class AsyncJdbcWrapper {

    private final Scheduler jdbcScheduler;

    @Autowired
    public AsyncJdbcWrapper(Scheduler jdbcScheduler) {

        this.jdbcScheduler = jdbcScheduler;
    }

    public <T> Mono<T> async(Callable<T> callable) {
        return Mono.fromCallable(callable)
                .subscribeOn(jdbcScheduler)
                .publishOn(Schedulers.parallel());
    }
}

然后使用它来包装 jdbc 调用，如下所示:

Mono<Integer> userIdMono = asyncWrapper.async(() -> userDao.getUserByUUID(request.getUserId()))
                .map(userOption -> userOption.map(u -> u.getId())
                        .orElseThrow(() -> new IllegalArgumentException("Unable to find user with ID " + request.getUserId())));

我有两个问题:

1) 我是否正确地将阻塞调用的执行推送到另一组线程？作为对这些东西相当陌生的人，我正在努力解决 subscribeOn()/publishOn() 的复杂性。

2) 假设我想利用生成的 mono，例如使用 userIdMono 的结果调用 API，它将在哪个调度程序上执行？是专门为 jdbc 调用创建的线程，还是 Reactor 通常在其中运行的主(？)线程？例如

userIdMono.map(id -> someApiClient.call(id));

Answer 1

1) 使用 subscribeOn 可以正确地将 JDBC 工作放在 jdbcScheduler

上

2) 两者都不是，Callable 的结果 - 在 jdbcScheduler 上计算时，是 publishOn 的 parallel 调度程序，因此您的 map 将在 Schedulers.parallel() 池中的线程上执行(而不是占用 jdbcScheduler)。