如何有条件地执行 Reactor 的 Mono/Flux 供应商算子？答案

【问题标题】：How to conditionally execute Reactor's Mono/Flux supplier operator?如何有条件地执行 Reactor 的 Mono/Flux 供应商算子？
【发布时间】：2019-08-28 23:32:01
【问题描述】：

如何有条件地执行then(Mono<T>)操作符？

我有一个返回Mono<Void> 的方法。它还可以返回错误信号。我想使用then 运算符（或任何其他运算符），只有在前一个操作完成且没有错误信号时。

谁能帮我找到合适的供应商运营商？

    public static void main(String[] args) {
        Mono.just("GOOD_SIGNAL")//It can also be a BAD_SIGNAL
                .flatMap(s -> firstMethod(s))
                .then(secondMethod())
                .subscribe()
        ;
    }

    private static Mono<String> secondMethod() {
        //This method call is only valid when the firstMethod is success
        return Mono.just("SOME_SIGNAL");
    }

    private static Mono<Void> firstMethod(String s) {
        if ("BAD_SIGNAL".equals(s)) {
            Mono.error(new Exception("Some Error occurred"));
        }

        return Mono
                .empty()//Just to illustrate that this function return Mono<Void>
                .then();
    }

-谢谢

【问题讨论】：

标签： java spring-webflux project-reactor

【解决方案1】：

then 在其源中传播错误，因此它涵盖了这一方面。据我了解，您不能使用flatMap 而不是then 的原因是由于firstMethod()，源是空的。在这种情况下，结合defer() 和then() 来实现您所寻求的懒惰：

Mono.just("GOOD_SIGNAL")//It can also be a BAD_SIGNAL
    .flatMap(s -> firstMethod(s))
    .then(Mono.defer(this::secondMethod))
    .subscribe();

【讨论】：

为什么要用这种方式defer？
它将阻止方法2的急切执行，只有在源代码首先完成时才运行它

【解决方案2】：

首先，我想强调一下，Reactor 的 Mono/Flux（接下来会考虑 Mono）具有以下条件运算符（至少是我所知道的）：

Mono#switchIfEmpty
Mono#defaultIfEmpty
Mono#filter 和其他一些供应商运营商的组合（例如Mono#flatMap）

第二点是Mono#then：

忽略此 Mono 中的元素并将其完成信号转换为提供的 Mono 的发射和完成信号。在生成的 Mono 中重放错误信号。

所以这意味着then 即将返回值（空的或提供的），不管它之前是什么。

考虑到所有这些，您的解决方案将如下所示：

public static void main(String[] args) {
        Mono.just("GOOD_SIGNAL")//It can also be a BAD_SIGNAL
                .flatMap(s -> firstMethod(s))
                .switchIfEmpty(secondMethod())
                .doOnError(...)//handle you error here
                .subscribe();
    }

private static Mono<String> secondMethod() {
     //This method call is only valid when the firstMethod is success
     return Mono.just("SOME_SIGNAL");
}

private static Mono<Void> firstMethod(String str) {
    return Mono.just(str)
               .filter(s -> "BAD_SIGNAL".equals(s))
               .map(s -> new Exception("Some Error occurred: "+s))
               .flatMap(Mono::error);
}

【讨论】：