弹簧反应堆:单声道.zip在空单声道上失败

Question

我正在使用Spring Reactor 3.1.0.M3，并且有一个用例，我需要合并来自多个来源的Mono。我发现如果其中一个 Monos 是空的 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时，在响应为空时跳过运算器。当我删除 m3 时，它都按预期工作并返回"AB"。 有没有办法通过检测空的单声道来处理这个问题？ 另外，有没有办法让组合器方法知道对象的类型，而不必强制转换？

Answer 1

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.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();
});

编辑

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

如果其中一个单声道是空的单声道，则 zip 将失败而不会出错

和

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

空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
});

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

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

Answer 2

在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 3

Mono.zip

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

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

Answer 4

应该Mono.zip：

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

此外，您可以通过Mono.just(Optional.empty())，而不是禁止null或Mono.empty()通过zip()。

Answer 5

完全披露，我的例子基本上是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。因此，要使其完整，您需要有一个非 null 值，在这种情况下为空字符串，但是数据类型的新实例供您稍后在代码中检查也很好