如何在 RxJava 2 中使用 onResume() 和 onPause() 方法订阅和取消订阅？答案

【问题标题】：How can I subscribe and unsubscribe using onResume() and onPause() methods with RxJava 2?如何在 RxJava 2 中使用 onResume() 和 onPause() 方法订阅和取消订阅？
【发布时间】：2017-11-11 15:12:46
【问题描述】：

嗯，我需要将发射绑定到我的活动的生命周期。我怎样才能做到这一点？什么时候应该创建观察者和可观察的实例？

【问题讨论】：

标签： rx-android rx-java2 activity-lifecycle

【解决方案1】：

如果您有一个 observable 需要有时而不是其他时间的数据，有一种简单的方法可以使用 switchMap() 运算符订阅和取消订阅。

假设你有一个 observable 想要来自：

Observable<LocationData> locationDataObservable;

那么，如果你引入一个switching observable：

PublishSubject<Boolean> switchObservable = PublishSubject.create();

您可以控制对第一个 observable 的订阅：

Observable<LocationData> switchedLocationDataObservable =
  switchObservable
    .switchMap( abled -> abled ? locationDataObservable : Observable.never() )
    .subscribe();

要启用接收数据，请执行

switchObservable.onNext( Boolean.TRUE );

并禁用，

switchObservable.onNext( Boolean.FALSE );

switchMap() 运算符会为您处理订阅和取消订阅。

【讨论】：

【解决方案2】：

不用 Observable.never() 也可以解决。只需将 BehaviourSubject 与 takeUntil 和 repeatWhen 运算符一起使用即可。

private fun getPauseUpdatesSource(): Flowable<Unit> {
    return setUpdatesPausedSubject
        .filter { it }
        .map { }
        .toFlowableLatest()
}

private fun getResumeUpdatesSource(): Flowable<Unit> {
    return setUpdatesPausedSubject
        .filter { it.not() }
        .map { }
        .toFlowableLatest()
}

在 rx 链中：

locationDataObservable
                .takeUntil(getPauseUpdatesSource())
                .repeatWhen { it.switchMap { getResumeUpdatesSource() } }

它只是暂停你的链并且不会发出任何空的 observables

【讨论】：