java - 在 RxJava 中组合未知数量的 Observable

Question

我有一种情况，我必须根据请求创建 N Observable 对象。我知道我可以使用 zip 来组合已知数量的 Observable。但是，我无法理解如何组合未知数量的 Observable。

我不清楚的部分是传递给 zip 的函数。根据 Observable 的数量，我必须创建一个带有 N 个参数的 lambda。

所有的 Observable 返回不同类型的对象。

Answer 1

更新:

我最终采用了解决并行请求列表的正常方法。只需使用 flatMap、merge、zip 和任何组合 rx 运算符。

我们唯一需要特别做的是对每个请求使用 .subscribeOn(Schedulers.io())。所有其他事情，发送并行请求或同时发送将由 rxjava 完美安排。

如果你想看到效果，请尝试以下操作:

private void runMyTest() {
    List<Single<String>> singleObservableList = new ArrayList<>();
    singleObservableList.add(getSingleObservable(500, "AAA"));
    singleObservableList.add(getSingleObservable(300, "BBB"));
    singleObservableList.add(getSingleObservable(100, "CCC"));
    Single.merge(singleObservableList)
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(System.out::println); }

private Single<String> getSingleObservable(long waitMilliSeconds, String name) {
    return Single
            .create((SingleOnSubscribe<String>) e -> {
                    try {
                        Thread.sleep(waitMilliSeconds);
                    } catch (InterruptedException exception) {
                        exception.printStackTrace();
                    }
                    System.out.println("name = " +name+ ", waitMilliSeconds = " +waitMilliSeconds+ ", thread name = "
+Thread.currentThread().getName()+ ", id =" +Thread.currentThread().getId());
                    if(!e.isDisposed()) e.onSuccess(name);
                })
            .subscribeOn(Schedulers.io()); }

输出:

System.out: name = CCC, waitMilliSeconds = 100, thread name = RxCachedThreadScheduler-4, id =463

System.out: CCC

System.out: name = BBB, waitMilliSeconds = 300, thread name = RxCachedThreadScheduler-3, id =462

System.out: BBB

System.out: name = AAA, waitMilliSeconds = 500, thread name = RxCachedThreadScheduler-2, id =461

System.out: AAA

//******以前的答案但不准确*******//

使用这个；解决了我的问题:

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

示例:

public Observable<CombinedData> getCombinedObservables() {
        List<Observable> observableList = new ArrayList<>();

        observableList.add(observable1);
        observableList.add(observable2);
        observableList.add(observable3);
        observableList.add(observable4);

        return Observable.zip(observableList, new Function<Object[], CombinedData>() {
            @Override
            public CombinedData apply(Object[] objects) throws Exception {
                return new CombinedData(...);
            }
        });
    }