android - RxJava 结合多个可观察对象的弹性方式

Question

我有多个返回 Observable 的模块:O1、O2、O3...On所有模块的结果应该合并到一个可观察的 Ocomb 中，这样单个任务可能会失败，但组合不会被终止或受到单个问题的影响。在我当前的解决方案中，我遇到了以下示例中的各种问题:

此代码结合了我的模块的输出:

public Observable<Data> getModuleData(){
    List<Observable<Data>> tasks = new ArrayList<>();
    for(MyModule module : modules){
        tasks.add(module.getData());
    }
    return Observable
            .mergeDelayError(Observable.from(tasks))
            .onBackpressureBuffer(MAX_BUFFER)
            .observeOn(AndroidSchedulers.mainThread());
}

现在，我想显示属性 X 例如'name' 所有发出的数据对象:

public List<String> getNames() {
    return  getModuleData() 
            .map(new Func1<Data, String>() {
                @Override
                public String call(Data data) {
                    return data.getName();
                }
            })
            .timeout(600, TimeUnit.MILLISECONDS)
            .toList()
            .toBlocking()
            .firstOrDefault(new ArrayList<String>());
}

getNames() 方法应该返回一个列表，因此会阻止执行。

问题 1似乎有一个 issue在 RxJava 中，如果我调用 observeOn() 并使其阻塞，无论超时等是什么，它都不会返回。

问题 2如果 onObserve() 被删除，代码将工作，但在应用程序的不同位置，我在 UI 中呈现非阻塞可观察对象的结果。将显示数据，但之后我的 UI 会做一些疯狂的事情。每次数据更改时，我都必须触摸我的 UI 列表组件来刷新屏幕。

问题 3某些模块可能会产生内部错误或不会调用 onCompleted()。我认为 mergeDelayError() 和 timeout() 的组合可以处理这些情况并为无响应的模块调用 onCompleted()。但是，如果其中一个模块不调用 onCompleted() 并且删除了 timeout() 语句，则阻塞调用将永远不会返回。

问题:

什么是组合多个 observable 的最佳方式，这样单个 observable 可能会失败，但它被处理为 onCompleted()/被忽略并且不会影响组合的 observable？

在不停止执行或结束循环的情况下进行组合的可观察阻塞和处理超时的最佳解决方案是什么？

Answer 1

What is the best way to combine multiple observable so that individual observables can fail but it's handled as onCompleted() / ignored and does not affect the combined observable?

          Observable.from(modules)
            .flatMap(MyModule::getData)
                .onErrorResumeNext(Observable.empty())
                .timeout(600,TimeUnit.MILLISECONDS, Observable.empty())
            .toList()
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(System.out::println);

通过在 subscribe() 之前添加 toBlocking 可以将上述流转换为阻塞，但除了测试之外没有多大意义