.zip 方法中的 Observable 数量是否有限制？答案

【问题标题】：Is There A Limit to The Number of Observables in .zip Method?.zip 方法中的 Observable 数量是否有限制？
【发布时间】：2018-06-19 21:40:58
【问题描述】：

在 Kotlin 的 zip 方法中用作参数的 Observable 的数量似乎是有限制的。如果这是准确的，那么最好的选择是什么？

例如，当我使用 9 个参数时，它按预期工作。当我添加第 10 个参数时，我收到错误 无法推断此参数的类型。请明确说明

Observable.zip(
            //TODO: parameterize exchange symbols based on pair
            methodOne() as Observable<Any>),
            methodTwo() as Observable<Any>),
            methodThree() as Observable<Any>),
            methodFour() as Observable<Any>),
            methodFive() as Observable<Any>),
            methodSix() as Observable<Any>),
            methodSeven() as Observable<Any>),
            methodEight() as Observable<Any>),
            methodNine() as Observable<Any>),
            { oneResult, twoResult, threeResult, fourResult, fiveResult, sixResult, sevenResult, eightResult, nineResult ->
                    //logic here applying computation to results
            })
            .subscribe(
                    {},
                    {
                        println(String.format("Error: %s", it.message))
                    })
            .unsubscribe()
}

【问题讨论】：

标签： java android kotlin rx-java rx-kotlin

【解决方案1】：

RxJava 最多只支持 9 个不同的源代码 zip。除此之外，您必须使用zip(Iterable<ObservableSource>, Func<Object[],R>) 方法并将Object[] 的每个元素转换回其各自的类型。

返回一个Observable，它发出指定combiner 函数的结果，该函数应用于按顺序由其他ObservableSources 中的Iterable 发出的项目组合。 zip 以严格的顺序应用此函数，因此新 ObservableSource 发出的第一项将是该函数应用于每个源 ObservableSources 发出的第一项的函数的结果；新的ObservableSource 发出的第二项将是应用于每个ObservableSources 发出的第二项的函数的结果；等等。

从 zip 返回的结果 ObservableSource<R> 将调用 onNext 的次数与发出最少项目的源 ObservableSource 的 onNext 调用次数一样多。

如果其中一个源比其他源短，则操作员按指定的顺序订阅其源，并在处置其他源时急切地完成。因此，那些其他来源可能永远无法运行完成（因此不会调用doOnComplete()）。如果源的长度完全相同，也会发生这种情况；如果源 A 完成且 B 已被消耗且即将完成，则操作员检测到 A 不会发送进一步的值，它将立即处理 B。例如：
zip(Arrays.asList(range(1, 5).doOnComplete(action1), range(6, 5).doOnComplete(action2)), (a) -> a)
action1 将被调用，但 action2 不会。要解决此终止属性，请同时使用doOnDispose(Action) 或使用using() 在完成或dispose() 调用的情况下进行清理。关于方法签名的注意事项：由于 Java 不允许使用新的 T[] 创建泛型数组，因此此运算符的实现必须创建一个 Object[]。不幸的是，传递给该方法的Function<Integer[], R> 会触发ClassCastException。

【讨论】：

【解决方案2】：

根据 Oracle Java 文档：http://reactivex.io/RxJava/javadoc/rx/Observable.html#zip(java.lang.Iterable,%20rx.functions.FuncN)

这些方法存在：

static <R> Observable<R>    zip(java.lang.Iterable<? extends Observable<?>> ws, FuncN<? extends R> zipFunction)

static <R> Observable<R>    zip(Observable<?>[] ws, FuncN<? extends R> zipFunction)

static <R> Observable<R>    zip(Observable <? extends Observable<?>> ws, FuncN<? extends R> zipFunction)

static <T1,T2,R> Observable<R>  zip(Observable<? extends T1> o1, Observable<? extends T2> o2, Func2<? super T1,? super T2,? extends R> zipFunction)

static <T1,T2,T3,R>
Observable<R>   zip(Observable<? extends T1> o1, Observable<? extends T2> o2, Observable<? extends T3> o3, Func3<? super T1,? super T2,? super T3,? extends R> zipFunction)

static <T1,T2,T3,T4,R>
Observable<R>   zip(Observable<? extends T1> o1, Observable<? extends T2> o2, Observable<? extends T3> o3, Observable<? extends T4> o4, Func4<? super T1,? super T2,? super T3,? super T4,? extends R> zipFunction)

static <T1,T2,T3,T4,T5,R>
Observable<R>   zip(Observable<? extends T1> o1, Observable<? extends T2> o2, Observable<? extends T3> o3, Observable<? extends T4> o4, Observable<? extends T5> o5, Func5<? super T1,? super T2,? super T3,? super T4,? super T5,? extends R> zipFunction)


static <T1,T2,T3,T4,T5,T6,R>
Observable<R>   zip(Observable<? extends T1> o1, Observable<? extends T2> o2, Observable<? extends T3> o3, Observable<? extends T4> o4, Observable<? extends T5> o5, Observable<? extends T6> o6, Func6<? super T1,? super T2,? super T3,? super T4,? super T5,? super T6,? extends R> zipFunction)

static <T1,T2,T3,T4,T5,T6,T7,R>
Observable<R>   zip(Observable<? extends T1> o1, Observable<? extends T2> o2, Observable<? extends T3> o3, Observable<? extends T4> o4, Observable<? extends T5> o5, Observable<? extends T6> o6, Observable<? extends T7> o7, Func7<? super T1,? super T2,? super T3,? super T4,? super T5,? super T6,? super T7,? extends R> zipFunction)

static <T1,T2,T3,T4,T5,T6,T7,T8,R>
Observable<R>   zip(Observable<? extends T1> o1, Observable<? extends T2> o2, Observable<? extends T3> o3, Observable<? extends T4> o4, Observable<? extends T5> o5, Observable<? extends T6> o6, Observable<? extends T7> o7, Observable<? extends T8> o8, Func8<? super T1,? super T2,? super T3,? super T4,? super T5,? super T6,? super T7,? super T8,? extends R> zipFunction)

static <T1,T2,T3,T4,T5,T6,T7,T8,T9,R>
Observable<R>   zip(Observable<? extends T1> o1, Observable<? extends T2> o2, Observable<? extends T3> o3, Observable<? extends T4> o4, Observable<? extends T5> o5, Observable<? extends T6> o6, Observable<? extends T7> o7, Observable<? extends T8> o8, Observable<? extends T9> o9, Func9<? super T1,? super T2,? super T3,? super T4,? super T5,? super T6,? super T7,? super T8,? super T9,? extends R> zipFunction)

所以如果有超过 9 个元素使用数组或可迭代对象会更简单

【讨论】：

【解决方案3】：

是的，Observable.zip() 有几个重载方法，ObservableSource 参数的最大数量是 9：

public static <T1, T2, T3, T4, T5, T6, T7, T8, T9, R> Observable<R> zip(
        ObservableSource<? extends T1> source1, ObservableSource<? extends T2> source2, ObservableSource<? extends T3> source3,
        ObservableSource<? extends T4> source4, ObservableSource<? extends T5> source5, ObservableSource<? extends T6> source6,
        ObservableSource<? extends T7> source7, ObservableSource<? extends T8> source8, ObservableSource<? extends T9> source9,
        Function9<? super T1, ? super T2, ? super T3, ? super T4, ? super T5, ? super T6, ? super T7, ? super T8, ? super T9, ? extends R> zipper) {

如果您想拥有超过 9 个来源，请查看 zipArray 或 zipIterable

【讨论】：