Rxjava关于Disposable你应该知道的事

关于disposable

Disposable类

• dispose():主动解除订阅
• isDisposed():查询是否解除订阅 true 代表 已经解除订阅

rxjava虽然好用，但是总所周知，容易遭层内存泄漏。也就说在订阅了事件后没有及时取阅，导致在activity或者fragment销毁后仍然占用着内存，无法释放。而disposable便是这个订阅事件，可以用来取消订阅。但是在什么时候取消订阅呢？我知道有两种方式:

• 使用CompositeDisposable
  看源码，CompositeDisposable的介绍很简单

A disposable container that can hold onto multiple other disposables and offers O(1) add and removal complexity.

一个disposable的容器，可以容纳多个disposable，添加和去除的复杂度为O(1)。
这里需要注意的是在该类的addAll方法有这么一句注释

Atomically adds the given array of Disposables to the container or disposes them all if the container has been disposed

也就是说，如果这个CompositeDisposable容器已经是处于dispose的状态，那么所有加进来的disposable都会被自动切断。

所以说可以创建一个BaseActivity，用CompositeDisposable来管理订阅事件disposable，然后在acivity销毁的时候，调用compositeDisposable.dispose()就可以切断所有订阅事件，防止内存泄漏。

CompositeDisposable类

1、可以快速解除所有添加的Disposable类.
2、每当我们得到一个Disposable时就调用CompositeDisposable.add()将它添加到容器中, 在退出的时候, 调用CompositeDisposable.clear() 即可快速解除.

案例代码如下:

CompositeDisposable compositeDisposable=new CompositeDisposable();
Observable.create(new ObservableOnSubscribe<Integer>() {
          @Override
          public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
              emitter.onNext(1);
              emitter.onComplete();或者 emitter.onError(new Throwable("O__O "));
          }
      }).subscribe(new Observer<Integer>() {
          private Disposable mDisposable;
          @Override
          public void onSubscribe(Disposable d) {
              <!-- 订阅   -->
              mDisposable = d;
              <!-- 添加到容器中 -->
              compositeDisposable.add(d);
          }
          @Override
          public void onNext(Integer value) {
        <!-- 判断mDisposable.isDisposed()如果解除了则不需要处理 -->
          }
          @Override
          public void onError(Throwable e) {
          }
          @Override
          public void onComplete() {
          }
      });
      <!-- 解除所有订阅者 -->
compositeDisposable.clear();

• 在oError和onComplete后调用disposable.dispose();，也就是上面我给的例子中的方法。
  查看源码，ObservableCreate的静态类CreateEmitter就是这种方式实现的。同时也可以看到，onError和onComplete不可以同时调用的原因：每次掉用过onError或onComplete其中一个方法后，就会掉用dispose()方法，此时订阅取消，自然也就不能掉用另一个方法了

new Observer<Integer>() {
          private Disposable mDisposable;
          private int i;

          @Override
          public void onSubscribe(Disposable d) {
              Log.d(TAG, "subscribe");
              mDisposable = d;
          }

          @Override
          public void onNext(Integer value) {
              Log.d(TAG, "onNext: " + value);
              i++;
              if (i == 2) {
                  Log.d(TAG, "dispose");
                  mDisposable.dispose();
                  Log.d(TAG, "isDisposed : " + mDisposable.isDisposed());
              }
          }

          @Override
          public void onError(Throwable e) {
              Log.d(TAG, "error");
          }

          @Override
          public void onComplete() {
              Log.d(TAG, "complete");
          }
      }