一、关于 Disposable
任何订阅者模式的代码，都需要注意注册与注销的配对出现，否则会出现内存泄漏。RxJava2 提供了 Disposable（ RxJava1 中是 Subscription），在适当时机取消订阅、截断数据流。当在 Android 中使用时尤其要注意，避免内存泄露。
private CompositeDisposable compositeDisposable = new CompositeDisposable();
@Override
public void onCreate() {
compositeDisposable.add(backendApi.loadUser()
.subscribe(this::displayUser, this::handleError));
}
@Override public void onDestroy() {
compositeDisposable.clear();
}
上面例子展示了在 Activity 等 LifecycleOwner 中的一般做法：使用 CompositeDisposable 收集所有的 Disposable 句柄，而后在 onDestroy 中调用 clear 统一注销。
clear 最终调用的是各个 Disposable 的 dispose 方法：
public interface Disposable {
void dispose();
boolean isDisposed();
}
当然，除了手动调用 dispose，也有一些自动框架可供使用， 如 RxLifecycle 、uber 的 AutoDispose 等， 但最终都要调用到 Disposable 的 dispose() 。
二、dispose 实现原理
Disposable disposable = Observable.create((ObservableOnSubscribe<Integer>) observableEmitter -> {
for (int i = 1; i <= 3; i++) {
observableEmitter.onNext(i);
}
})
.takeUntil(integer -> integer < 3)
.subscribe();
...
...
//在适当的是调用dispose取消订阅
disposable.dispose()
1、Disposable 是一个 Observer
调用 Observable.subscribe(…) 返回的 Disposable 本质是一个 LambdaObserver
public final Disposable subscribe(
@NonNull Consumer<? super T> onNext,
@NonNull Consumer<? super Throwable> onError,
@NonNull Action onComplete) {
LambdaObserver<T> ls = new LambdaObserver<>(
onNext, onError, onComplete,
Functions.emptyConsumer());
subscribe(ls);
return ls; //return as a Disposable
}
LambdaObserver 集众多接口于一身
public final class LambdaObserver<T> extends AtomicReference<Disposable> implements Observer<T>, Disposable
- 首先，是一个 Observer，被subscribe()后，通过onNext发射数据；
- 其次，是一个 Disposable，对外提供 dispose 方法；
- 最后，通过 AtomicReference，确保 dispose 线程安全的执行
@Override
public void dispose() {
DisposableHelper.dispose(this);
}
public static boolean dispose(AtomicReference<Disposable> field) {
Disposable current = field.get();
Disposable d = DISPOSED;
if (current != d) {
current = field.getAndSet(d);
if (current != d) {
if (current != null) {
current.dispose();
}
return true;
}
}
return false;
}
原子地设置 DISPOSED, 确保 AtomicReference 中的 Disposable 的 dispose 一定被调用，有且仅有一次。
2、onSubscribe 中传递 Disposable
AtomicReference 的 value 是在 Observer.onSubscribe 中被赋值的：
@Override
public void onSubscribe(Disposable d) {
if (DisposableHelper.setOnce(this, d)) { //设置 value
try {
onSubscribe.accept(this);
} catch (Throwable ex) {
...
}
}
}
那么 Observer.onSubscribe 又是何时被调用呢？
RxJava 的操作符都是一个 Observable 实现。操作符链式调用的本质就创建 Observable 并通过 subscribe 依次订阅。 subscribe 内部会用 subscribeActual ，这是每个操作符都必须实现的方法。
看一下 Observabel.create 的 subscribeActual ：
调用 Observer.onSubscrie()， 将当前 Disposable 作为 parent 传递给下游
protected void subscribeActual(Observer<? super T> observer) {
CreateEmitter<T> parent = new CreateEmitter<>(observer); // CreateEmitter是一个Diposable
observer.onSubscribe(parent); // Observer.onSubscrie()
try {
source.subscribe(parent);
} catch (Throwable ex) {
...
}
}
3、Observer 关联上下游
除 create、subscribe 这样的终端操作符以外，大部分的操作符的 Observer 结构如下：
/** The downstream subscriber. */
protected final Observer<? super R> downstream;
/** The upstream subscription. */
protected Disposable upstream;
public final void onSubscribe(Disposable d) {
...
this.upstream = d;
downstream.onSubscribe(this);
...
}
public void dispose() {
upstream.dispose();
}
- Observer 持有上下游对象：upstream 和 downstream
- onSubscribe 向下递归调用
- dipose 向上递归调用
在链式订阅中，向下游订阅 Observer 的同时，也关联了上游的 Disposable(Observer)
我们在最下端调用 subscribe 时，调用链上的 Observer 会建立上下游关联，当我们在下游调用 dispose 时，最终会递归调用到顶端（create）的 dispose
4、再看takeUntil的例子
根据上述分析，在回顾一下最初 takeUntil 的例子。
前面说过所有的操作符都是 Observable：
- takeUntil 对应的Observable： ObservableTakeUntilPredicate;
- create 对应的Observable： ObservableCreate
subscribe 调用链如下：
随着 onSubscribe 的调用，Disposable 也建立了如下引用链：
当我们调用 dispose 方法时，通过引用链递会最终调用到 CreateEmitter 的 dispose。
由于 CreateEmitter 将 AtomicReference 的 value 设为 DISPOSED
后续，onNext 中判断状态，当为 DISPOSED 时，数据流停止发射
@Override
public void onNext(T t) {
if (!isDisposed()) { //是否为DISPOSED
observer.onNext(t);
}
}