在 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：名称 = CCC，waitMilliSeconds = 100，线程名称 = RxCachedThreadScheduler-4, id =463

System.out: CCC

System.out：名称 = BBB，waitMilliSeconds = 300，线程名称 = RxCachedThreadScheduler-3, id =462

System.out: BBB

System.out：名称 = AAA，waitMilliSeconds = 500，线程名称 = 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(...);
            }
        });
    }

Answer 2

您可以让您的 Observables 返回常见类型的对象，将其合并并作为 List 处理：

class Result1 implements Result
class Result2 implements Result
class REsult3 implements Result

for(Observable o : yourObservableList)
    resultObservable.mergeWith(o) //use concat() if you need serial execution

resultObservable
    .toList()
    .doOnNext(results -> {
        //process your results (List<Result>)
    }
    .subscribe(...)

Answer 3

当我不得不根据应用程序中的用户选择对不同数量的 EditTexts 进行表单验证时，我遇到了同样的问题。

在我的具体情况下，所有添加的字段都必须包含内容。

这最终对我有用。

        val ob1 = RxTextView.textChanges(field1).skip(1)
        val ob2 = RxTextView.textChanges(field2).skip(1)
        val ob3 = RxTextView.textChanges(field3).skip(1)

        val observableList = arrayListOf<Observable<CharSequence>>()

        observableList.add(ob1)
        observableList.add(ob3)

        val formValidator = Observable.combineLatest(observableList, {
            var isValid = true
            it.forEach {
                val string = it.toString()
                if (string.isEmpty()) {
                    isValid = false
                }
            }
            return@combineLatest isValid
        })

        formValidator.subscribe { isValid ->
            if (isValid) {
                //do something
            } else {
                //do something
            }
        }