Spring Reactor：Mono.zip 在空 Mono 上失败

Question

我正在使用 Spring Reactor 3.1.0.M3 并且有一个用例，我需要合并来自多个来源的 Mono。我发现如果其中一个 Mono 是空的 Mono，则 zip 会失败而不会出现错误。

例子：

Mono<String> m1 = Mono.just("A");
Mono<String> m2 = Mono.just("B");
Mono<String> m3 = Mono.empty();

Mono<String> combined = Mono.zip(strings -> {
    StringBuffer sb = new StringBuffer();
    for (Object string : strings) {
        sb.append((String) string);
    }
    return sb.toString();
}, m1, m2, m3);
System.out.println("Combined " + combined.block());

添加 m3 时，在响应为 null 时跳过组合子。当我删除 m3 时，一切都按预期工作，并返回“AB”。 有没有办法通过检测空的 Mono 来处理这个问题？ 另外，有没有办法让组合器方法知道对象的类型而不必强制转换？

Answer 1

完全公开，我的示例基本上是 Brian Clozel 的，但稍作修改以突出显示其他数据类型。

Mono<String> m1 = Mono.just("A");
Mono<String> m2 = Mono.just("B");
Mono<String> m3 = Mono.fromSupplier(() -> new String());

Mono<String> combined = Mono.zip(m1, m2, m3).map(t -> {
    StringBuffer sb = new StringBuffer();
    sb.append(t.getT1());
    sb.append(t.getT2());
    sb.append(t.getT3());
    return sb.toString();
});

也就是说，zip 将忽略一个为空的条目，即 Mono.empty。所以为了完成它，你需要有一个非空值，在这种情况下是一个空字符串，但是一个新的数据类型实例供你稍后在代码中检查也是好的

Answer 2

应该是Mono.zip:

Mono<String> myStrings = Mono.zip(monoA, monoB)
  .map(tuple -> {
    return new StringBuffer()
      .append(t.getT1());
      .append(t.getT2());
});

也可以通过zip()而不是禁止null或Mono.empty()传递Mono.just(Optional.empty())。

Answer 3

在String 的情况下，很容易为空的情况定义一个默认值，这可以很好地解决问题，如 Brian 的回答中所述。但是，对于其他自定义类型，可能由于某种原因很难创建空对象。这些情况的替代方法是使用Optional。不过，这个解决方案有一些繁重的样板。

Mono<Optional<String>> m1 = Mono.just("A").map(Optional::of).defaultIfEmpty(Optional.empty());
Mono<Optional<String>> m2 = Mono.just("B").map(Optional::of).defaultIfEmpty(Optional.empty());
Mono<Optional<String>> m3 = Mono.<String>empty().map(Optional::of).defaultIfEmpty(Optional.empty());

Mono<String> combined = Mono.zip(strings -> {
    StringBuffer sb = new StringBuffer();
    for (Object string : strings) {
        ((Optional<String>) string).ifPresent(sb::append);
    }
    return sb.toString();
}, m1, m2, m3);
System.out.println("Combined " + combined.block());

Answer 4

Mono.zip

任何源的错误或空完成将导致其他源被取消，并且生成的 Mono 分别立即出错或完成。

当一个 Mono 源完成但没有价值时，如果您仍想继续执行其他源，那么 Mono.zipDelayError 是您的最佳选择。

Answer 5

zip 运算符的行为不是这样的。这实际上是违反直觉的：您的代码需要一个包含 3 个元素的元组，而您只得到两个？！？

在这种情况下，您可以控制，如果没有提供默认值，只有您可以决定什么是好的默认值（请记住，null 值被反应流规范禁止）。

Mono<String> m1 = Mono.just("A");
Mono<String> m2 = Mono.just("B");
Mono<String> m3 = Mono.empty().defaultIfEmpty("");

Mono<String> combined = Mono.when(m1, m2, m3).map(t -> {
    StringBuffer sb = new StringBuffer();
    sb.append(t.getT1());
    sb.append(t.getT2());
    sb.append(t.getT3());
    return sb.toString();
});

编辑

您似乎对Publisher 类型的性质感到困惑，请参阅：

如果其中一个 Mono 是空 Mono，则 zip 会失败且不会出现错误

和

因此，如果我尝试压缩 Mono 的文件，但由于某种原因其中一个是空的，则压缩会失败，我似乎无法输入任何代码来防止这种情况发生

空的Mono 不是失败案例：只是没有发出任何值并且成功完成。您可以通过更改代码示例来验证：

    combined.subscribe(
            s -> System.out.println("element: " + s), // doesn't execute
            s -> System.out.println("error: " + s), // doesn't execute
            () -> { System.out.println("complete!"); // prints
    });

因此，根据您的要求，您可以：

• 在这 3 个 Mono 实例上应用 defaultIfEmpty 运算符，如果有方便的默认值可以依赖
• 在组合Mono 上应用defaultIfEmpty 运算符，使用默认值，甚至将其转换为带有combined.switchIfEmpty(Mono.error(...)) 的错误消息