RxJava AutoDispose原理解析

版权声明：本文为博主原创文章，未经博主允许不得转载https://blog.csdn.net/wsygyb/article/details/90523082

概述

最近的项目采用AutoDispose解决RxJava内存泄漏的问题.相对于让组件继承RxActivity或者RxFragment,使用AutoDispose只需要简单地加上.`as`(AutoDispose.autoDisposable(AndroidLifecycleScopeProvider.from(this))),就能非常漂亮并且保持较少侵入性的解决内存泄漏问题.当然本篇文章的重点不在于讲述AutoDispose如何使用,而是阐述AutoDispose如何通过监听Lifecycle来实现dispose.

必备的知识体系

在讲述原理之前,需要确保具有以下知识点的储备：

1. Lifecycle ，通过实现LifecycleObserver的方式订阅Lifecycle的事件,订阅者不会强引用lifecycle本身；
2. RxJava2，了解Observable的链式引用原理.以及map、filter等操作符;

Lifecycle监听生命周期

通过<kbd>@OnLifecycleEvent</kbd>注解可以订阅Lifecycle的特定事件，同时订阅者不会强引用Lifecycle本身,从源码来看:比如SupportActivity,在调用getLifecycle时返回的是LifecycleRegistry,里面维持的是对Activity本身的弱引用.

public class MyObserver implements LifecycleObserver {
    @OnLifecycleEvent(Lifecycle.Event.ON_RESUME)
    public void onResume() {
    }
    @OnLifecycleEvent(Lifecycle.Event.ON_PAUSE)
    public void onPause() {
    }
    @OnLifecycleEvent(Lifecycle.Event.ON_DESTROY)
    public void onDestroy() {
    }    
}
aLifecycleOwner.getLifecycle().addObserver(new MyObserver());

RxJava的内存泄漏场景

当RxJava在异步线程中执行耗时任务(比如网络请求、IO等),在该任务未结束前由于持有Activity的Context将导致Activity无法被正常销毁。从代码来看,Observable.create创建了一个ObservableCreate(继承自Observable),该Observable在被订阅时会创建CreateEmitter来保存observer的实例.由于该实例引用Context,将使得Activity无法被回收.

    Observable.create<String>{
        observer -> 
         observer.onNext("")
         observer.onComplete()
     }
     .observeOn(Schedulers.io())
     .subscribe{
         t->
         val context=this@DActivity
         Thread.sleep(5000)
         Log.d("",context.toString())
     }

AutoDispose在Activity::onDestroy时避免内存泄漏

• AutoDispose在内部创建了ArchLifecycleObserver,采用<kbd>Event.ON_ANY</kbd>注解监听Lifecyc的生命周期
• 当Lifecycle发出ON_DESTROY事件时,ArchLifecycleObserver转发该事件给特定observer,该observer通过filter限定<kbd>Event.ON_DESTROY</kbd>事件通过
• 随后当Activity销毁时,Lifecycle发送事件给``ArchLifecycleObserver··,并调用<kbd>onDispose</kbd>方法取消对Lifecycle的监听。最后回调至ObservableCreate在订阅时创建的CreateEmitter的dispose方法,将CreateEmitter本身赋值为DISPOSED,销毁observer实例.

AutoDispose的创建

首先,AutoDispose会获取当前Context的lifecycle,并对应地创建一个observer用于监听lifecycle的变化。随后会 例如下面产生的一个序列图:

AutoDispose的创建流程
<center>图1:AutoDispose的创建流程</center>

首先,AutoDispose会调用当前LifecycleOwner的getLifecycle,实现如下所示:

  public static AndroidLifecycleScopeProvider from(LifecycleOwner owner) {
    return from(owner.getLifecycle());
  }

在本文中,LifecycleOwner的实例是Activity,故调用getLifecycle返回的是LifecycleRegistry.随后调用from(lifecycle, DEFAULT_CORRESPONDING_EVENTS)创建AndroidLifecycleScopeProvider,如下所示:

  public static AndroidLifecycleScopeProvider from(
          Lifecycle lifecycle,
          Function<Lifecycle.Event, Lifecycle.Event> boundaryResolver) {
    return new AndroidLifecycleScopeProvider(lifecycle, boundaryResolver);
  }

DEFAULT_CORRESPONDING_EVENTS是AndroidLifecycleScopeProvider内部创建的一个Function,该Function的作用用于返回对应初始生命周期的结束生命周期,声明如下所示:

  private static final Function<Lifecycle.Event, Lifecycle.Event> DEFAULT_CORRESPONDING_EVENTS =
      new Function<Lifecycle.Event, Lifecycle.Event>() {
        @Override public Lifecycle.Event apply(Lifecycle.Event lastEvent) throws Exception {
          switch (lastEvent) {
            case ON_CREATE:
              return Lifecycle.Event.ON_DESTROY;
            case ON_START:
              return Lifecycle.Event.ON_STOP;
            case ON_RESUME:
              return Lifecycle.Event.ON_PAUSE;
            case ON_PAUSE:
              return Lifecycle.Event.ON_STOP;
            case ON_STOP:case ON_DESTROY:
            default:
            ...
          }}};

随后,AndroidLifecycleScopeProvider内部根据传入的lifecycle创建了一个observer,用于监听本文中Activity的生命周期,并对应地创建一个Observable用于监听lifecycle的变化,实现如下:

class LifecycleEventsObservable extends Observable<Event> {

  private final Lifecycle lifecycle;
  private final BehaviorSubject<Event> eventsObservable = BehaviorSubject.create();

  @SuppressWarnings("CheckReturnValue") LifecycleEventsObservable(Lifecycle lifecycle) {
    this.lifecycle = lifecycle;
  }

  Event getValue() {
    return eventsObservable.getValue();
  }

  /**
   * Backfill if already created for boundary checking. We do a trick here for corresponding events
   * where we pretend something is created upon initialized state so that it assumes the
   * corresponding event is DESTROY.
   */
  void backfillEvents() {
    @Nullable Lifecycle.Event correspondingEvent;
    switch (lifecycle.getCurrentState()) {
      case INITIALIZED:
        correspondingEvent = ON_CREATE;
        break;
      case CREATED:
        correspondingEvent = ON_START;
        break;
      case STARTED:
      case RESUMED:
        correspondingEvent = ON_RESUME;
        break;
      case DESTROYED:
      default:
        correspondingEvent = ON_DESTROY;
        break;
    }
    eventsObservable.onNext(correspondingEvent);
  }

  @Override protected void subscribeActual(Observer<? super Event> observer) {
    ArchLifecycleObserver archObserver =
        new ArchLifecycleObserver(lifecycle, observer, eventsObservable);
    observer.onSubscribe(archObserver);
    if (!isMainThread()) {
      observer.onError(
          new IllegalStateException("Lifecycles can only be bound to on the main thread!"));
      return;
    }
    lifecycle.addObserver(archObserver);
    if (archObserver.isDisposed()) {
      lifecycle.removeObserver(archObserver);
    }
  }

  static final class ArchLifecycleObserver extends MainThreadDisposable
      implements LifecycleObserver {
    private final Lifecycle lifecycle;
    private final Observer<? super Event> observer;
    private final BehaviorSubject<Event> eventsObservable;

    ArchLifecycleObserver(Lifecycle lifecycle, Observer<? super Event> observer,
        BehaviorSubject<Event> eventsObservable) {
      this.lifecycle = lifecycle;
      this.observer = observer;
      this.eventsObservable = eventsObservable;
    }

    @Override protected void onDispose() {
      lifecycle.removeObserver(this);
    }

    @OnLifecycleEvent(Event.ON_ANY) void onStateChange(LifecycleOwner owner, Event event) {
      if (!isDisposed()) {
        if (!(event == ON_CREATE && eventsObservable.getValue() == event)) {
          // Due to the INITIALIZED->ON_CREATE mapping trick we do in backfill(),
          // we fire this conditionally to avoid duplicate CREATE events.
          eventsObservable.onNext(event);
        }
        observer.onNext(event);
      }
    }
  }
}

AutoDispose被订阅

本文中的示例代码如下:

Observable.create<String> {
     observer ->
     observer.onNext("first")
     observer.onNext("second")
 }
 .`as`(AutoDispose.autoDisposable(AndroidLifecycleScopeProvider.from(this)))
 .subscribe {
     it->
     val context=this@MainActivity
     Log.d("",context.toString())
 }

as操作符以及AutoDispose的订阅流程如下所示:

在这里插入图片描述
<center>图2:AutoDispose的订阅流程</center>

首先,Observable::create会创建一个ObservableCreate对象,而AutoDispose.autoDisposable会创建AutoDisposeConverter对象.as操作符调用前面生成的AutoDisposeConverter的apply方法,由于这里传入的是Observable对象,故执行下列方法调用:

 public ObservableSubscribeProxy<T> apply(Observable<T> upstream) {
     //这里的scope对应的是之前调用deferredResolvedLifecycle返回的Maybe对象
     return (ObservableSubscribeProxy)upstream.to(new ObservableScoper(scope));
 }

Observable::to调用ObservableScoper的apply(observable)方法创建一个ObservableSubscribeProxy对象,声明如下:

public ObservableSubscribeProxy<T> apply(final Observable<? extends T> observableSource)
      throws Exception {
    return new ObservableSubscribeProxy<T>() {
      @Override public Disposable subscribe() {
        return new AutoDisposeObservable<>(observableSource, scope()).subscribe();
      }

      @Override public Disposable subscribe(Consumer<? super T> onNext) {
        //observableSource -> 示例中通过Observable::create创建的ObservableCreate对象
        //scope() -> 调用deferredResolvedLifecycle生成的Maybe<LifecycleEndNotification>
        return new AutoDisposeObservable<>(observableSource, scope()).subscribe(onNext);
      }
    }
}

待ObservableSubscribeProxy创建完成后,as操作符调用完成。接着正式进入订阅流程。通过Consumer订阅,执行AutoDisposeObservable的subscribeActual方法,如下:

@Override protected void subscribeActual(Observer<? super T> observer) {
  //source -> 实例中创建的ObservablCreate对象
  source.subscribe(new AutoDisposingObserverImpl<>(scope, observer));
}

创建AutoDisposingObserverImpl实例,并接着调用该实例的onSubscribe、onNext方法.

final class AutoDisposingObserverImpl<T> extends AtomicInteger implements AutoDisposingObserver<T> {
    //lifecycleDisposable -> 
    private final AtomicReference<Disposable> lifecycleDisposable = new AtomicReference<>();
    
    AutoDisposingObserverImpl(Maybe<?> lifecycle, Observer<? super T> delegate) {
      //lifecycle -> Maybe<LifecycleEndNotification>
      this.lifecycle = lifecycle;
      //delegate实际上为observer,因为在AutoDisposingObserverImpl代理了Observable的调用
      //所以此处命名为deletagete
      this.delegate = delegate;
    }
      
    @Override public void onSubscribe(final Disposable d) {
      //创建一个Observer对象
      DisposableMaybeObserver<Object> o = new DisposableMaybeObserver<Object>() {
        @Override public void onSuccess(Object o) {
          lifecycleDisposable.lazySet(AutoDisposableHelper.DISPOSED);
          AutoDisposableHelper.dispose(mainDisposable);
        }
    
        @Override public void onError(Throwable e) {
          lifecycleDisposable.lazySet(AutoDisposableHelper.DISPOSED);
          AutoDisposingObserverImpl.this.onError(e);
        }
    
        @Override public void onComplete() {
          lifecycleDisposable.lazySet(AutoDisposableHelper.DISPOSED);
          // Noop - we're unbound now
        }
      };
      //设置lifecycleDisposable为前面创建的Observer对象o
      if (AutoDisposeEndConsumerHelper.setOnce(lifecycleDisposable, o, getClass())) {
        //代理delegate的onSubscribe方法回调
        delegate.onSubscribe(this);
        //调用Maybe<LifecycleEndNotification>::subscribe方法
        lifecycle.subscribe(o);
        AutoDisposeEndConsumerHelper.setOnce(mainDisposable, d, getClass());
      }
    }
}

在AutoDisposingObserverImpl::onSubscribe的实现中,主要完成了以下事情:

• 创建了DisposableMaybeObserver对象并赋值给成员lifecycleDisposable
• 代理consumer的onSubscribe方法调用
• 执行<kbd>Maybe<LifecycleEndNotification></kbd>的订阅

在ScopeUtil::deferredResolvedLifecycle方法中生成的Maybe<LifecycleEndNotification>对象如下:

Maybe.defer(new Callable<MaybeSource<? extends LifecycleEndNotification>>() {
      @Override public MaybeSource<? extends LifecycleEndNotification> call() throws Exception {
      //provider -> AndroidLifecycleScopeProvider
      //lastEvent -> 对应LifecycleOwner的当前生命周期
      E lastEvent = provider.peekLifecycle();
      E endEvent;
      try {
        //provider -> AndroidLifecycleScopeProvider
        //provider.correspondingEvents() -> DEFAULT_CORRESPONDING_EVENTS
        //endEvent -> 匹配当前生命周期的结束生命周期,如ON_CREATE对应ON_DESTROY
        endEvent = provider.correspondingEvents()
            .apply(lastEvent);
      } catch (Exception e) {
        if (checkEndBoundary && e instanceof LifecycleEndedException) {
          Consumer<? super OutsideLifecycleException> handler
              = AutoDisposePlugins.getOutsideLifecycleHandler();
          if (handler != null) {
            handler.accept((LifecycleEndedException) e);
            return Maybe.just(LifecycleEndNotification.INSTANCE);
          } else {
            throw e;
          }
        } else {
          return Maybe.error(e);
        }
      }
      //provider.lifecycle() -> LifecycleEventsObservable
      //resolveScopeFromLifecycle -> 生成最终订阅的Maybe<LifecycleEndNotification>对象,
      //同时会对发生的事件进行过滤:skip(1) -> 跳过第一个数据,map -> 只匹配与endEvent相同的事件
      return resolveScopeFromLifecycle(provider.lifecycle(), endEvent);
    }
});

Maybe.defer只有在订阅之后才会创建相应的MaybeSource对象.在订阅时首先调用call回调,主要完成以下任务:
1.调用AndroidLifecycleScopeProvider::peekLifecycle方法,获取当前lifecycle的生命周期lastEvent
2.调用provider.correspondingEvents().apply(lastEvent)获取对应lastEvent的结束生命周期endEvent
3.生成<kbd>Maybe<LifecycleEndNotification></kbd>对象,该对象对LifecycleEventsObservable进行包装之后再交由DisposableMaybeObserver订阅

我们先来看看LifecycleEventsObservable的实现:

class LifecycleEventsObservable extends Observable<Event> {
    private final Lifecycle lifecycle;
    @Override protected void subscribeActual(Observer<? super Event> observer) {
        ArchLifecycleObserver archObserver =
            new ArchLifecycleObserver(lifecycle, observer, eventsObservable);
        observer.onSubscribe(archObserver);
        if (!isMainThread()) {
          .....
        }
        //此处对Lifecycle进行了监听.
        lifecycle.addObserver(archObserver);
        if (archObserver.isDisposed()) {
          lifecycle.removeObserver(archObserver);
        }
    }
    
    //用于监听生命周期的Observer子类
    static final class ArchLifecycleObserver extends MainThreadDisposable
      implements LifecycleObserver {
      private final Lifecycle lifecycle;
      private final Observer<? super Event> observer;
      private final BehaviorSubject<Event> eventsObservable;

      ArchLifecycleObserver(Lifecycle lifecycle, Observer<? super Event> observer,
            BehaviorSubject<Event> eventsObservable) {
        this.lifecycle = lifecycle;
        this.observer = observer;
        this.eventsObservable = eventsObservable;
      }

      @Override protected void onDispose() {
        lifecycle.removeObserver(this);
      }
      //当生命周期发生变化时,会回调此函数
      @OnLifecycleEvent(Event.ON_ANY) void onStateChange(LifecycleOwner owner, Event event) {
        if (!isDisposed()) {
          //此处过滤掉重复的CREATE event
          if (!(event == ON_CREATE && eventsObservable.getValue() == event)) {
            // Due to the INITIALIZED->ON_CREATE mapping trick we do in backfill(),
            // we fire this conditionally to avoid duplicate CREATE events.
            eventsObservable.onNext(event);
          }
          //observer -> DisposableMaybeObserver对象
          observer.onNext(event);
        }
      }
  }

}

随后看看resolveScopeFromLifecycle函数对LifecycleEventsObservable进行了什么操作:

public static <E> Maybe<LifecycleEndNotification> resolveScopeFromLifecycle(
    Observable<E> lifecycle,
    final E endEvent) {
  return lifecycle.skip(1)
      .map(new Function<E, Boolean>() {
        @Override public Boolean apply(E e) throws Exception {
          return e.equals(endEvent);
        }
      })
      .filter(IDENTITY_BOOLEAN_PREDICATE)
      .map(TRANSFORM_TO_END)
      .firstElement();
}

    首先,skip掉了Observable发处的第一个数据,因为这通常对应LifecycleOwner的初始化生命周期,所以并不会影响进行dispose的时机。随后,筛选出与endEvent一致的事件让其通过,比如在onCreate进行订阅则endEvent为ON_DESTROY.最后对事件名称进行了下映射并返回满足条件的第一个事件.
    这里需要对firtstElement操作符进行说明,调用firtstElement封装后的Observable子类会对下发的数据进行计数。当计数到达特定下标时,这里对应0,则认为数据派发已经完成随后调用s.dispose().对应的代码实现如下:

public void onNext(T t) {
    if (done) {
        return;
    }
    long c = count;
    if (c == index) {
        done = true;
        //上流的Observable对象
        s.dispose();
        //actual -> observer对象,这里指代DisposableMaybeObserver实例
        actual.onSuccess(t);
        return;
    }
    count = c + 1;
}

s为上流的Observable对象,由于rxjava支持链式调用,故每个Observable都会以引用的方式引用上流的Observable对象.s.dispose会回调至LifecycleEventsObservable::onDispose方法取消对lifecycle的监听.随后,actual.onScucess会回调至DisposableMaybeObserver::onSuccess方法,如下:

final class AutoDisposingObserverImpl<T> extends AtomicInteger implements AutoDisposingObserver<T> {
    DisposableMaybeObserver<Object> o = new DisposableMaybeObserver<Object>() {
      @Override public void onSuccess(Object o) {
        //销毁observer本身
        lifecycleDisposable.lazySet(AutoDisposableHelper.DISPOSED);
        //mainDisposable -> 在本文中对应由Observable.create创建的ObservableCreate实例,此处进行销毁
        AutoDisposableHelper.dispose(mainDisposable);
      }
    }
}

在onSuccess回调中,主要做了两件事:

• 销毁自身对应的DisposableMaybeObserver实例
• 销毁上游的Observable对象,由于AutoDisposingObserverImpl作为observer保存在上游的Observable中,在调用Observable::dispose时会同时销毁observer成员.故AutoDispose完成了在生命周期结束时对Observable和Observer相应的实例销毁,从而避免了内存泄漏.