RxJava 结合 Observables答案

【问题标题】：RxJava Combining ObservablesRxJava 结合 Observables
【发布时间】：2017-09-07 23:52:10
【问题描述】：

我希望将多个可观察对象的结果组合成一个父可观察对象。我不能一次访问所有的 observables，它们将在程序的整个执行过程中被订阅。

到目前为止，这是我的方法：

...
compoundObservable = compoundObservable.mergeWith(firstObservable);
...
compoundObservable = compoundObservable.mergeWith(secondObservable);
...

这种方法没有奏效，因为otherObservable 发送的事件没有被compoundObservable 的订阅者注册。

如何组合这些 observables？

【问题讨论】：

在您的情况下可以选择使用Subject 吗？
所以，firstObservable 和 secondObservable 只有在有人订阅（当时是什么）compoundOversable 之后才会创建？
firstObservable 和 secondObservable 已创建。我希望在compoundObservable 中累积每个可观察的结果。至于Subject，我愿意接受 - 发布答案！
好的，但是你什么时候订阅compoundObservable？关键是所有的 Observable 都是不可变的：mergeWith 返回一个与compoundObservable 之前的值不同的新实例，即使您使用相同的变量名来保留它。要点是：您必须在订阅之前merge(With) - 如果您之后合并，您将永远不会使用合并后的 Observable。

标签： java rx-java reactive-programming observable

【解决方案1】：

使用Subject，您可以随时订阅可观察对象，并且该主题的所有订阅者都将获得所有项目。不过，这在调试时可能有点棘手。

PublishSubject<Object> subject = PublishSubject.create();
Observable<Object> observable1 = ...;
observable1.subscribe(subject);

subject.subscribe(...) // will eventually also get emisions from observable2

Observable<Object> observable2 = ...;
observable2.subscribe(subject);

【讨论】：

【解决方案2】：

如果我正确理解了您的问题，您可以使用多个运算符来配对您的 observable：

现在您应该确定您的 observables 以何种速率产生数据，以及您计划如何将它们产生的数据配对。

例如：

CombineLatest - 拥有 obs1 和 obs2，如果 obs1 以比 obs2 更快的速度生成数据，则来自 obs1 的数据将与 obs2 生成的最新数据配对。因此，来自 obs1 的多个项目与来自 obs2 的相同项目配对。
Zip - 有 obs1 和 obs2，如果 obs1 产生数据的速度比 obs2 快，那么只有当 obs2 产生值时，来自 obs1 的数据才会配对。
And/Then/When - 方便组合两个以上的 observables

注意：LINK 上还有一本免费书籍（适用于 .NET - RxNET）。查看组合序列中的配对序列部分以获得更深入的解释和示例。语法相似，并且运算符的目的在不同的实现中保持不变。

【讨论】：