project reactor中的包装阻塞I/O

Question

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

在对这个主题进行了一些阅读之后，我希望将阻塞数据库调用的执行与我的非阻塞异步代码的其余部分分开。

我定义了一个专用的jdbcScheduler：

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

@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());
    }
}

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())));

2）假设我想利用得到的mono，例如调用带有userIdMono结果的API，它将在哪个调度器上执行？是专门为jdbc调用创建的调度器，还是reactor通常在其中操作的主（？）线程？例如。

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

Answer 1

1）使用subscribeon可以正确地将JDBC工作放在JDBCScheduler上

2）都不是，当callable的结果在jdbcScheduler上计算时，是发布在并行调度器上，因此您的映射将在调度器.parallel（）池的线程上执行（而不是占用jdbcScheduler)。