用 RxJava 替换 EventBus - N 个订阅者总是在监听答案

【问题标题】：Replacing EventBus with RxJava - N subscribers always listening用 RxJava 替换 EventBus - N 个订阅者总是在监听
【发布时间】：2014-09-15 02:24:20
【问题描述】：

我在 Android 应用中将 EventBus 模式替换为 RxJava。我有事件提醒任何感兴趣的各方更新我的缓存单例中的数据。任何时候调用 Web 服务，数据都会更新，并且订阅者会通过发布的事件收到警报。

我在 RxJava 中使用AsyncSubject 进行了类似的设置。观察者从主题中得到一个事件，然后他们得到一个onComplete 事件并取消订阅。这在 UI 第一次加载时起作用，但是当需要刷新数据时，不会通知订阅者。我如何告诉Subscribers 继续监听来自Subject 的更多onNext 事件？

我需要一个Subject 来报告最新项目。 PublishSubject 仅在订阅后发出项目，因此不太符合我的需求。我的订阅者在不同的时间开始观察（可能在第一个数据事件之后），所以我需要主题发出观察到的最后一个项目，然后为后续项目保持流打开。似乎我需要 AsyncSubject 和 PublishSubject 的组合。有没有办法通过内置类来实现这一点，还是我需要创建自己的主题？

WebServiceObservable OR CacheObservable
                  ^
                  |
             AsyncSubject
                  ^
                  |
                /   \
              /       \
            /           \
      UiObserver1   UiObserver2

【问题讨论】：

标签： android rx-java

【解决方案1】：

BehaviorSubject 将满足您的需求。 https://github.com/Netflix/RxJava/wiki/Subject#behaviorsubject

如果您需要更复杂的行为，您可以随时编写自己的 Subject 实现。这样做似乎很简单。

【讨论】：

我最终使用了BehaviorSubject，但我使用内部观察者仅在父 Observable 调用 onNext() 时调用，忽略任何 onComplete 事件以保持流打开。更多信息：groups.google.com/forum/#!topic/rxjava/2bNl14lIjRo
在您的情况下忽略 onComplete 是必要的。我应该在我的回答中提到这一点，但这在我发布的链接中有明确的解释。

【解决方案2】：

答案有点晚，但比 BehaviorSubject 更好的方案可能是来自 RxRelay lib 的 BehaviorRelay。当您需要不同的行为但希望在所有模块之间共享单点交互时，也可以使用更多全局解决方案RxHub

【讨论】：

【解决方案3】：

我认为如果将 BehaviorSubject 与 switchOnNext 运算符一起使用会更简单。

switchOnNext( ) 将发出 Observables（本例中为 BehaviorSubject）的 Observable 转换为单个 Observable，该 Observable 发出这些 Observables 中最近发出的项目

switchOnNext( ) 返回的 Observable 取消订阅之前发出的 Observable 开始从最新的 Observable 发出项目

public class PhotoModel{
  BehaviorSubject<Observable<Photo>> subject = BehaviorSubject.create(...);

  public void setUserId(String id){
   subject.onNext(Api.getUserPhoto(photoId));
  }

  public Observable<Photo> subscribeToPhoto(){
    return Observable.switchOnNext(subject);
  }
}

When should one use RxJava Observable and when simple Callback on Android?

【讨论】：