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

问题描述

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

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

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

解决方法

将阻塞调用包装在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的结果。

,

为此，最好使用.then()运算符：

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

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