doFinally 是否在 Reactor 中的同一线程上执行答案

【问题标题】：Does doFinally execute on the same thread in ReactordoFinally 是否在 Reactor 中的同一线程上执行
【发布时间】：2020-10-29 16:58:15
【问题描述】：

doFinally 是否在同一个线程上执行？下面的代码会阻塞主线程吗？

mono
.map(fileName -> asyncDownloadFile(fileName, folderName))
.doFinally(v -> {
    FileUtils.cleanDirectory(folderName); // this method is blocking
});

如果是这样，在 doFinally 的单独线程中执行 cleanDirectory 的最佳方法是什么？

【问题讨论】：

标签： java spring-webflux project-reactor reactor

【解决方案1】：

将阻塞调用包装在 Runnable 中并在单独的 thread 上运行它：

Runnable task = () -> {FileUtils.cleanDirectory(folderName)};

Mono<Object> cleanDirPromise = Mono.fromRunnable(task);

mono
.map(fileName -> asyncDownloadFile(fileName, folderName))
.doFinally(v -> {
    cleanDirPromise.subscribeOn(Schedulers.parallel()).subscribe();
});

注意：这本质上是一个即发即弃的调用，您不会真正关心cleanDirPromise 的结果。

【讨论】：

Schedulers.parallel() 创建一个固定的工作线程池，其内核数量与正在运行的系统一样多。 Schedulers.boundedElastic() 是更好的选择 creates new worker pools as needed and reuses idle ones. Worker pools that stay idle for too long (the default is 60s) are also disposed. ... This is a better choice for I/O blocking work. Schedulers.boundedElastic() is a handy way to give a blocking process its own thread so that it does not tie up other resources. 取自文档 projectreactor.io/docs/core/release/reference/#schedulers
Schedulers.parallel() 应该用于需要在多核上进行计算的高 CPU 密集型任务。
和subscribe 应该被删除没有理由在应用程序中间订阅。
@ThomasAndolf doFinally 不订阅内部流。外部订阅与内部流订阅无关。就调度程序而言，这不是问题的一部分。但是，是的，任何阻塞任务都需要在弹性上运行。
doFinally 返回一个 Mono，所以不需要订阅 projectreactor.io/docs/core/release/api/reactor/core/publisher/…

【解决方案2】：

为此最好使用.then()操作符：

mono
    .map(fileName -> asyncDownloadFile(fileName, folderName))
    .then()
    .flatMap(
        Mono.fromRunnable(() -> FileUtils.cleanDirectory(folderName))
            .subscribeOn(Schedulers.boundedElastic())
    )
    ...

运算符then() 保证cleanDirectory 将在asyncDownloadFile 之后执行，还允许您构建一个管道并处理错误。

【讨论】：

您还需要将处理程序添加到错误中，但最好在 doFinally 中运行新订阅。