在 rxJava 中压缩超过 9 个可观察量

Question

面对需要进行 10-12 个小型并行查询并组合结果的情况。但是，如果有一种zip方法可以让您组合多达9个可观察量，那么如何做更多的事情我不明白。我尝试使用zip方法

public static <T, R> Observable<R> zip(Iterable<? extends ObservableSource<? extends T>> sources, Function<? super Object[], ? extends R> zipper)

但得到一个错误

java.lang.ClassCastException: java.util.ArrayList cannot be cast to io.reactivex.ObservableSource

尝试如下所示

List<Observable> list = new ArrayList<>();
list.add(orderRepository.getOne(54, "id"));
list.add(orderRepository.getTwo(54, "id"));
list.add(orderRepository.getThree(54, "id"));
list.add(orderRepository.getFour(54, "id"));
list.add(orderRepository.getFive());
list.add(orderRepository.getSix(54, "id"));
list.add(orderRepository.getSeven(54, "id"));
list.add(orderRepository.getEight());
list.add(orderRepository.getNine());
list.add(orderRepository.getTen(54, "id"));
list.add(orderRepository.getEleven(54, "id"));
Observable.fromIterable(list);
return Observable.zip(list,
new Function<Object[], OrderModel>() {
@Override
public OrderModel apply(Object[] objects) throws Exception {
Logger.trace("objects = ", objects);
return new OrderModel();
}
});

请举例说明如何在 Java 或 Kotlin 中执行此操作。

Answer 1

Observable one = Observable.zip(
orderRepository.getOne(54, "id"), 
orderRepository.getTwo(54, "id"),
// Etc up to five (I think)
Function zipper
);
Observable two = Observable.zip(
one, orderRepository.getSix(54, "id"),
orderRepository.getSeven(54, "id"),
// Etc up to five,
Function zipper
);

冲洗并重复，直到将所有可观察内容压缩在一起。

Answer 2

下面是如何将 Observable#zip 重载与 Iterable 一起使用的示例。你会得到一个新的 Object[]，因为 Java 泛型不支持新的 T[]。

在此示例中，您将看到 T 的可观察量列表，该列表将在 Observable#zip 中使用。zipper函数会将每个对象强制转换为字符串并返回T列表。因此>你会得到可观察的。

此外，我建议您确保给定的可观察列表不是空的。当一个空的可观察量列表被提供给 Observable#zip 时，它将立即完成而不会发出任何内容(感谢 @akarnokd(。或者你可以使用 Observable#switchIfEmpty 来提供一个回退值(例如空列表(

@Test
void x() {
Observable<String> z1 = Observable.just("1");
Observable<String> z2 = Observable.just("2");
Observable<String> z3 = Observable.just("3");
List<Observable<String>> observables = Arrays.asList(z1, z2, z3);
Observable<List<String>> zip =
Observable.zip(
observables,
objects -> {
List<String> resultList =
Stream.of(objects).map(o -> (String) o).collect(Collectors.toList());
return resultList;
});
zip.test()
.assertNoErrors()
.assertComplete()
.assertValueCount(1)
.assertValueAt(
0,
r -> {
assertThat(r).contains("1", "2", "3");
return true;
});
}

Answer 3

Observable.zip(Observable1 , Observable2 ,... Observable9  , 
Function9<return_value_observable1 , return_value_observable2 , .... , 
return_value_observable9 , ZipModel { ->
}