你是否了解 RxJava 的 Disposable ?

简介: 你是否了解 RxJava 的 Disposable ?

关于 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();

当调用 disposable.dispose(); 时,代码如何执行?

先卖个关子,文章最后揭晓答案

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 一定被调用,有且仅有一次。

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.createsubscribeActual

调用 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) {
        ...
    }
}

Observer 关联上下游

createsubscribe 这样的终端操作符以外,大部分的操作符的 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

再看takeUntil的例子

根据上述分析,在回顾一下最初 takeUntil 的例子。

前面说过所有的操作符都是 Observable:

  • takeUntil 对应的Observable: ObservableTakeUntilPredicate;
  • create 对应的Observable: ObservableCreate

subscribe 调用链如下:

image.png

当我们调用 dispose 方法时,通过引用链递会最终调用到 CreateEmitter 的 dispose。

由于 CreateEmitter 将 AtomicReference 的 value 设为 DISPOSED 后续,onNext 中判断状态,当为 DISPOSED 时,数据流停止发射

@Override
public void onNext(T t) {
    if (!isDisposed()) { //是否为DISPOSED
        observer.onNext(t);
    }
}

关于onComplete

通过下面的测试发现当 onComplete 调用后会,会自动调用 dispose。

@Test 
public void testDisposed(){
  boolean isDisposed = false;
    TestObserver<Integer> to = Observable.<Integer>create(subscriber -> {
        subscriber.setDisposable(new Disposable() {
            @Override
            public boolean isDisposed() {
                return isDisposed;
            }
            @Override
            public void dispose() {
                isDisposed = true;
            }
        });
        subscriber.onComplete();
    }).test();
    to.assertComplete();
    assertTrue(isDisposed);
}

果然,ObservableEmitteronComplete 中调用了 dispose:

 public void onComplete() {
            if (!isDisposed()) {
                try {
                    observer.onComplete();
                } finally {
                    dispose();
                }
            }
        }

关于内存泄漏

调用dipose确实可以终止数据流,但是不等于没有内存泄露。

查看 ObservableCreate 的源码可知,dispose只是简单地设置了 DISPOSED 状态,Observe 中关联的上下游对象并没有释放。所以当订阅了静态的 Observable 时,无法避免内存泄漏。

但是当订阅一个 Subject 时,dispose 确实可以有效释放对象,避免内存泄漏:

public void dispose() {
  if (super.tryDispose()) {
    parent.remove(this); //对象删除
  }
}

关于 dispose 的实时性

前面分析知道,对于终端操作符 createsubscribe 等,其 Observer 在 dispose 时会标记当前状态为 DISPOSED。但对于其他操作符的 dispose 只是递归向上调用 parent 的 dispose 而已,并没有 DISPOSED 状态的设置,也就不会拦截发射中的数据。

调用dispose后,RxJava数据流不一定会立即停止,大部分操作符在调用 dispose 后,数据依然会发射给下游

关于 dispose 的实时性测试,下文可供参考

medium.com/stepstone-t…

目录
相关文章
RxJava2 中 doFinally 和 doAfterTerminate 的比较
RxJava2 中 doFinally 和 doAfterTerminate 的比较
295 0
|
数据处理
RxJava2实现RxBus
RxJava2实现RxBus
174 0
|
安全 Android开发
详解 RxJava 的 Disposable
RxJava2 的 Disposable,可以在适当时机取消订阅、截断数据流,避免 Android 中的内存泄露。
1191 0
|
Java Go Android开发
RxJava2
函数式编程是一种编程范式。我们常见的编程范式有命令式编程、函数式编程和逻辑式编程。我们常见的面向对象编程是一种命令式编程。命令式编程是面向计算机硬件的抽象,有变量、赋值语句、表达式和控制语句。而函数式编程是面向数学的抽象,将计算描述为一种表达式求值,函数可以在任何地方定义,并且可以对函数进行组合。响应式编程是一种面向数据流和变化传播的编程范式,数据更新是相关联的。把函数式编程里的一套思路和响应式编程合起来就是函数响应式编程。函数响应式编程可以极大地简化项目,特别是处理嵌套回调的异步事件、复杂的列表过滤和变换或者时间相关问题。在Android开发中使用函数响应式编程的主要有两大框架:
169 0
RxJava2
|
设计模式 存储 Java
XTask与RxJava的使用对比
XTask与RxJava的使用对比
166 0
XTask与RxJava的使用对比
|
负载均衡 算法 Java
RxJava 并行操作
RxJava 并行操作
419 0
RxJava 并行操作
|
Java API
RxJava 之 ParallelFlowable
RxJava 之 ParallelFlowable
260 0
RxJava 之 ParallelFlowable
RxJava/RxAndroid : doOnNext
RxJava/RxAndroid : doOnNext import android.support.v7.app.AppCompatActivity; import android.
1075 1
|
Java
冷饭热炒——RxJava
已经超过一个月没有写文章了,原因无非就是工作太忙。最近终于恢复以前的节奏,任务开始正常了起来。忙里偷闲,写一写人们写烂了的RxJava。这篇文章主要分析RxJava事件的产生以及变化的原理,Ok,let's go! 0. 前言 本次源码分析使用的是RxJava2,版本2.1.14。
1070 0
|
Java 数据库 Android开发
06.RxJava初探
基本使用 在build.gradle中加入配置,注意,rxJava和rxAndroid版本一定要相互兼容,不然可能会报错More than one file was found with OS independent path 'META-INF/rxjava.
1182 0