了解数据流和多个订阅者（使用改造）答案

【问题标题】：Understanding streams of data and multiple subscribers (using retrofit)了解数据流和多个订阅者（使用改造）
【发布时间】：2015-02-08 00:50:06
【问题描述】：

假设我有 2 个 Observable（A 和 B），它们本质上是网络调用（使用 Retrofit 提供上下文）。

应用目前的流程如下：

A 和 B 大约同时启动（异步）。
B 在用户交互时执行 0 次或更多次

考虑到这 2 个 observables/api 调用，我有 3 个不同的场景需要监听。

我想立即知道 Observable A 何时完成
我想立即知道 Observable B 何时完成
我想知道两者什么时候完成

首先，这是 RxJava 的一个很好的用例吗？

我知道如何单独处理每个场景（最后使用zip），但我不知道如何同时处理所有场景。

如果我订阅 Observable A，A 就开始了。如果我订阅 B，则 B 开始。如果 A & B 在我订阅 zip(a, b) 之前完成，我可能会错过该事件并且永远不会真正看到它完成，对吗？

任何一般性指导将不胜感激。我的 RxJava 知识很薄：P

【问题讨论】：

标签： android retrofit rx-java rx-android

【解决方案1】：

您可以使用三种不同的 observable 来实现这一点，每种情况一个。

由于您必须在每个可观察对象之间共享状态，因此您必须将改造的冷可观察对象转换为热可观察对象。（有关此主题的更多信息，请参阅here）

ConnectableObservable a = service.callA().publish(); 
ConnectableObservable b = service.callB().publish();

a.subscribe((e) -> { /* onNext */ }, (ex) -> {/* onError */},  () -> {/* when A is completed */ });
b.subscribe((e) -> { /* onNext */ }, (ex) -> {/* onError */},  () -> {/* when B is completed */ });
a.mergeWith(b).subscribe((e) -> { /* onNext */ }, (ex) -> {/* onError */},  () -> {/* when A and B are completed */ });

a.connect(); // start subscription to a
b.connect(); // start subscription to b

不要在 onCompleted 方法之间共享对象，否则您将不得不处理并发问题。

【讨论】：

太棒了，这真的很有帮助。你介意扩展一下我可能遇到的并发问题吗？如果我在同一个线程上观察所有这些，这仍然是一个问题吗？我认为 A 或 B 将首先完成，然后是 A+B。
Observable a、Observable b 和 Observable ab（a 和 b 合并的 observable）可以在不同的线程中完成。因此，对 onCompleted 方法的调用不能是连续的（与 rx 合约相反）。您可以在 onCompleted 方法中读取名为“hello”的共享变量。下一行，您再次读取相同的变量。值可以改变，因为另一个可观察对象的另一个 onCompleded 方法可以改变它。