在RxJava中组合未知数量的Observable

Question

我有一种情况，我必须根据请求创建N Observable对象。据我所知，我可以使用zip来组合一定数量的Observables。但是，我试图理解如何组合未知数量的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(...);
            }
        });
    }

Answer 2

您可以使Observable返回常见类型的对象，将其合并并作为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
            }
        }