两个 Mono 与条件的组合答案

【问题标题】：Combination of two Mono with condition两个 Mono 与条件的组合
【发布时间】：2019-10-02 13:08:32
【问题描述】：

我想根据某些条件组合结果表两个Mono。 Mono 都是 WebClient 调用的结果：

第一个是单个呼叫，期望快速响应。
第二个是多个响应缓慢的呼叫的组合。

如果第一个结果满足某些条件，则“取消”第二个Mono 的想法是为了节省时间并避免不必要的网络调用。如果第一个的Mono 结果不够zip 它与第二个Mono。

一个 Kotlin 代码示例来解释我的想法：

fun getResult(): Mono<Result> {

    val trivialResultMono: Mono<Result> = webClient.getResult()

    val nonTrivialResultMono: Mono<Result> = webClient
            .getResult()
            .flatMap { webClient.getResult1(it) }
            .flatMap { webClient.getResult2(it) }
            .flatMap { webClient.getResult2(it) }

    //here I need to check if trivial result satisfies some condition,
    //for example trivialResult.size > 5 if it's true I just return
    //trivialResultMono from getResult() function, 
    //it it's false something like this:
    return Mono.zip(trivialResultMono, nonTrivialResultMono) { trivialResult, nonTrivialResult ->
        trivialResult + nonTrivialResult
    }

}

更新：

为了更清楚，假设 trivialResult 在 1 秒内出现，nonTrivialResult 在 2 秒内出现。如果是trivialResult.size > 5，我想在 1 秒内得到最终结果，否则在 2 秒内得到。

仅使用Mono.zip(trivialResultMono, nonTrivialResultMono)，我总是会在 2 秒内得到最终结果。

使用filter + switchIfEmpty，如果trivialResult.size > 5 需要1 秒，否则需要3 秒。如果我错了，请纠正我。

【问题讨论】：

鉴于第一次调用很快，难道不可以直接等待它完成，然后过滤掉不合适的结果并切换到第二次调用吗？
假设第二次调用比第一次长 1.5 倍。在这种情况下，这个解决方案并不是那么理想。

标签： reactive-programming spring-webflux project-reactor

【解决方案1】：

您可以过滤您的 trivialResultMono 并应用 switchIfEmpty 运算符

return trivialResultMono
        .filter(trivialResult -> trivialResult.size > 5)
        .switchIfEmpty(Mono.zip(...))

merge 方法的更新：

Mono<Result> zipResultMono = Mono.zip...

return Flux.merge(
        trivialResultMono.map(trivialResult -> Tuples.of(1, trivialResult)),
        zipResultMono.map(zipResult -> Tuples.of(2, zipResult)))
        .filter(tuple ->
                (tuple.getT1().equals(1) && tuple.getT2().size > 5) ||
                        tuple.getT1().equals(2))
        .next()
        .map(Tuple2::getT2);

如果 zipResult 的大小总是大于 5，您可以跳过转换为 Tuple2

【讨论】：

看起来像一个解决方案，但在 trivialResult.size zip 更长。
您可以使用trivialResult.cache()。对几乎所有系统而言，持续时间差异不应该是至关重要的。
我的意思是在这种情况下，对琐碎和非琐碎结果的请求将不会同时执行，而是以串行方式执行。
在这种情况下，您可以使用Flux.merge 然后过滤并获取第一个结果。我已经更新了答案。

【解决方案2】：

您可以通过flatMap 和map 实现此目的：

trivial.flatMap(trivialResult -> {
  if (trivialResult.size > 5) {
    return Mono.just(trivialResult);
  } else {
    return nonTrivial.map(nonTrivialResult -> trivialResult + nonTrivialResult);
  }
});

【讨论】：