RxJava这么好用却容易内存泄漏?解决办法是...

时间:2022-06-25
本文章向大家介绍RxJava这么好用却容易内存泄漏?解决办法是...,主要内容包括其使用实例、应用技巧、基本知识点总结和需要注意事项,具有一定的参考价值,需要的朋友可以参考一下。

/   简介   /

熟悉RxJava的同学,当我们开启一个异步任务时,通常需要在Activity/Fragment销毁时,及时关闭异步任务,否则就会有内存泄漏的。

一般的做法是订阅成功后,拿到Disposable对象,在Activity/Fragment销毁时,调用Disposable对象的dispose()方法,将异步任务中断,也就是中断RxJava的管道,代码如下:

Disposable disposable = Observable              
    .interval(0, 1, TimeUnit.SECONDS)  //开启一个定时器
    .subscribe(aLong -> {                       

    });                                         

//Activity/Fragment销毁时,中断RxJava管道     
if (disposable != null && !disposable.isDisposed()) {
    disposable.dispose();                            
} 

这种做法在代码的执行效率上是最高效、性能最优的,然而这种做法在开发效率上却是最低的。

试想,如果我们开启了n个异步任务,就需要在Activity/Fragment销毁时中断n个异步任务。对于这种写法,身患强迫症的我,实在不能接受。也许你们会说,可以使用CompositeDisposable类,就可以避免手写关闭n个异步任务的代码,只需要关闭一次即可。没毛病,确实可以,然而这种做法也仅仅是避免了我们手写关闭异步任务的代码而已。追求极致的我,也不能接受这种写法,此时我就想,能不能就用一行代码解决这个问题呢?于是乎,就开启了我的探索之路,于是乎,就有了RxLife。

先来介绍下RxLife,相较于trello/RxLifecycle、uber/AutoDispose,具有如下优势:

  • 直接支持在主线程回调
  • 支持在子线程订阅观察者
  • 简单易用,学习成本低
  • 性能更优,在实现上更加简单

gradle依赖

dependencies {
   implementation 'com.rxjava.rxlife:rxlife:1.0.8'
   //if you use AndroidX
   implementation 'com.rxjava.rxlife:rxlife-x:1.0.8'
} 

#/   RxLife使用   /

Activity/Fragment

首先,我们来看看在Activity/Fragment上如何使用,如下:

//在Activity/Fragment上
Observable.interval(1, 1, TimeUnit.SECONDS)
    .as(RxLife.as(this)) //这里this 为LifecycleOwner接口对象
    .subscribe(aLong -> {
        Log.e("LJX", "onNext aLong=" + aLong);
    }); 

没错,就是这么简单粗暴,在这,我们只需要将RxLife.as(this)传入RxJava的as操作符即可。此时当Activity/Fragment销毁,就会自动关闭RxJava管道,避免内存泄漏。

View

接着来看看在View上如何使用,如下:

//在View上
Observable.interval(1, 1, TimeUnit.SECONDS)  //隔一秒发送一条消息
    .as(RxLife.as(this)) //这里this 为View对象
    .subscribe(aLong -> {
        Log.e("LJX", "onNext aLong=" + aLong);
    }); 

咦?这跟上面的代码不是一模一样的吗?是的,代码一模一样,但是在这我们传入的this是一个View对象。此时当View从窗口中移除时(执行了onDetachedFromWindow方法),就会自动关闭RxJava管道,避免内存泄漏。

ViewModel

ViewModel是Google Jetpack里面的组件之一,由于它能自动感知Activity/Fragmeng的销毁,所以RxLife单独为它做了适配。在ViewModel中使用RxLife,需要继承RxLife的 ScopeViewModel 类,然后就可以跟上面一样,优雅的使用RxLife.as(this),如下:

public class MyViewModel extends ScopeViewModel {

    public MyViewModel() {
        Observable.interval(1, 1, TimeUnit.SECONDS)
            .as(RxLife.as(this)) //这里的this 为Scope接口对象
            .subscribe(aLong -> {
                Log.e("LJX", "onNext aLong=" + aLong);
            });
    }
} 

此时当Activity/Fragmeng销毁时,就会自动关闭RxJava管道,避免内存泄漏。

注意:要想ViewModel对象感知Activity/Fragment销毁事件,就不能使用new 关键字创建对象,必须要通过ViewModelProviders类获取ViewModel对象,如下:

//在Activity/Fragment上
MyViewModel viewModel = ViewModelProviders.of(this).get(MyViewModel.class) 

任意类

相信大家对MVP都非常的熟悉了,在P层,我们一般都有发送Http请求的需求, 此时,我们也希望,在Activity/Fragment销毁时,能自动将Http关闭,所以RxLife对任意类做了点适配工作。在任意类中,我们需要继承RxLife的BaseScope类,然后就优雅的使用RxLife.as(this)了,如下:

public class Presenter extends BaseScope {

    public Presenter(LifecycleOwner owner) {
        super(owner); 
        Observable.interval(1, 1, TimeUnit.SECONDS)
            .as(RxLife.as(this)) //这里的this 为Scope接口对象
            .subscribe(aLong -> {
                Log.e("LJX", "onNext aLong=" + aLong);
            });
    }
} 

kotlin

在上面的代码中,我们使用了as操作符,然后在kotlin中,as是一个关键字,使用起来就不是很方便,所以RxLife对kotlin做了适配工作,在kotlin中,我们可以使用life替代as操作符,并且更加的简洁,如下:

Observable.intervalRange(1, 100, 0, 200, TimeUnit.MILLISECONDS)     
    .life(this)                                                 
    .subscribe { aLong ->                                                    
        Log.e("LJX", "onNext aLong=" + aLong)
    } 

/   原理   /

说起原理,其实trello/RxLifecycle、uber/AutoDispose、RxLife三者的原理都是一样的,都是拿到最低层观察者的Disposable对象,然后在某个时机,调用该对象的Disposable.dispose()方法中断管道,以达到目的。原理都一样,然而实现却大不相同。

  • trello/RxLifecycle (3.0.0版本) 内部只有一个管道,但却有两个事件源,一个发送生命周期状态变化,一个发送正常业务逻辑,最终通过takeUntil操作符对事件进行过滤,当监听到符合条件的事件时,就会将管道中断,从而到达目的。
  • uber/AutoDispose(1.2.0版本) 内部维护了两个管道,一个是发送生命周期状态变化的管道,我们称之为A管道,另一个是业务逻辑的管道,我们称至为B管道,B管道持有A管道的观察者引用,故能监听A管道的事件,当监听到符合条件的事件时,就会将A、B管道同时中断,从而到达目的。
  • RxHttp 内部只有一个业务逻辑的管道,通过自定义观察者,拿到Disposable对象,暴露给Scope接口,Scope的实现者就可以在合适的时机调用Disposable.dispose()方法中断管道,从而到达目的。

RxLife具体实现

光从文字层面上所原理,好像有点抽象,接下来,我们看看RxLife在代码层面上是如何实现的。在上面的代码案例中,我们皆能看到RxLife.as(this)这行代码的身影,那这个as方法接收的是什么类型的参数呢?我们看看源码:

上面一共有10个as系列方法,其中有8个是对外提供的。而且前面9个方法最终都会调用第10个as(Scope scope, boolean onMain)方法。

我们先粗略来看几个方法

  • as(LifecycleOwnerowner owner) 方法,接收的是一个LifecycleOwner接口对象,简单介绍下这个接口,这个接口对象能使我们自定义的类感知Activity/Fragment的生命周期回调。我们常见的Activity/Fragment就实现了这个接口,所以我们就能够在Activity/Fragment中调用此。法。
  • as(View view) 这个方法就很直观了,直接接收一个View对象,我们在View上调用的就是这个方法。
  • as(Scope scope) 方法接收一个Scope接口对象,后面会对这个接口介绍,这里可以告诉大家的是,在上面的ViewModel及任意类中继承的ScopeViewModel、BaseScope类都实现了Scope接口,所以我们在ViewModel及任意类中调用的就是这个as方法。

Scope接口

Scope,翻译过来就是作用域的意思。那么什么是作用域,简单来说,就是一个对象从创建到死亡,这就是它的作用域,比如:Activity/Fragment的作用域就是从onCreate到onDestroy;View的作用域就是从onAttachedToWindow到onDetachedFromWindow;ViewModel的作用域就是从构造方法到onCleared方法;其它任意类的作用域就是从创建到销毁,当然,你也可以自己指定一些类的作用域。到这,我们来看看Scope接口里面都有啥:

public interface Scope {

    //订阅事件时,回调本方法,即在onSubscribe(Disposable d)方法执行时回调本方法
    void onScopeStart(Disposable d);

    //onError/onComplete 时调回调此方法,即事件正常结束时回调
    void onScopeEnd();
} 

此接口描述的就是RxJava的作用域,即从事件订阅到事件结束。到这,也许有人已经知道了,只要我们实现了这个接口,就能拿到Disposable对象,然后就可以在某个时刻,中断RxJava短道,提前结束RxJava作用域。从而使得RxJava的作用域小于等于调用者的作用域,避免了内存泄漏。

我们简单看一下BaseScope类的具体实现。

public class BaseScope implements Scope, GenericLifecycleObserver {

    private CompositeDisposable mDisposables;

    public BaseScope(LifecycleOwner owner) {
        owner.getLifecycle().addObserver(this);
    }

    @Override
    public void onScopeStart(Disposable d) {
        addDisposable(d);
    }

    @Override
    public void onScopeEnd() {}

    private void addDisposable(Disposable disposable) {
        CompositeDisposable disposables = mDisposables;
        if (disposables == null) {
            disposables = mDisposables = new CompositeDisposable();
        }
        disposables.add(disposable);
    }

    private void dispose() {
        final CompositeDisposable disposables = mDisposables;
        if (disposables == null) return;
        disposables.dispose();
    }

    @Override
    public void onStateChanged(LifecycleOwner source, Event event) {
        //Activity/Fragment 生命周期回调
        if (event == Event.ON_DESTROY) {  //Activity/Fragment 销毁
            source.getLifecycle().removeObserver(this);
            dispose(); //中断RxJava管道
        }
    }
} 

可以看到,BaseScope实现非常简单,在onScopeStart方法中拿到Disposable对象添加进CompositeDisposable对象,然后在Activity/Fragment销毁使,调用CompositeDisposable对象的dispose方法,统一中断RxJava管道,从而达到目的。

Scope一共有4个实现类,分别是:LifecycleScope、ViewScope、ScopeViewModel及BaseScope,BaseScope上面已经介绍,其它3个原理都一样,只是在实现上会有一点点不同,这就不在一一介绍了。

#/   问题暴露   /

我们知道,任意类想要监听Activity/Fragment生命周期回调,都必须要实现LifecycleObserver接口,然后通过以下代码添加进观察者队列

owner.getLifecycle().addObserver(this); 

这行代码的内部是通过FastSafeIterableMap类来管理观察者的,而这个类是非线程安全的,如下:

我们来看看trello/RxLifecycle、uber/AutoDispose、RxLife这三者是如何处理这个问题的。

trello/RxLifecycle 

RxLifecycle库是AndroidLifecycle类感知生命周期,简单看看源码:

public final class AndroidLifecycle implements LifecycleProvider<Lifecycle.Event>, LifecycleObserver {

    public static LifecycleProvider<Lifecycle.Event> createLifecycleProvider(LifecycleOwner owner) {
        return new AndroidLifecycle(owner);
    }

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

    private AndroidLifecycle(LifecycleOwner owner) {
        owner.getLifecycle().addObserver(this);
    }
    //中间省略部分代码
    @OnLifecycleEvent(Lifecycle.Event.ON_ANY)
    void onEvent(LifecycleOwner owner, Lifecycle.Event event) {
        lifecycleSubject.onNext(event);
        if (event == Lifecycle.Event.ON_DESTROY) {
            owner.getLifecycle().removeObserver(this);
        }
    }
} 

可以看到,RxLifecycle是在对象创建时添加观察者,且它没有做任何处理,如果你在子线程使用,就需要额外注意了,而且它只有在页面销毁时,才会移除观察者,试想,我们在首页一般都会有非常多的请求,而这每一个请求都会有一个AndroidLifecycle对象,我们想请求结束就要回收这个对象,然而,这个对象还是观察者队列里,就导致了没办法回收,如果我们不停下拉刷新、上拉加载更多,对内存就是一个挑战。

RxLifecycle还有一个弊端时,当Activity/Fragment销毁时,始终会往下游发送一个onComplete事件,这对于在onComplete事件中有业务逻辑的同学来说,无疑是致命的打击。

uber/AutoDispose  

AutoDispose库我们看LifecycleEventsObservable类,如下

class LifecycleEventsObservable extends Observable<Event> {

  //省略部分代码
  @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);
    }
  }
  //省略部分代码 

可以看到,AutoDispose是在事件订阅时添加观察者,并且当前非主线程时,直接抛出异常,也就说明使用AutoDispose不能在子线程订阅事件。在移除观察者方面,AutoDispose会在事件结束或者页面销毁时移除观察者,这一点要优于RxLifecycle。

RxLife

RxLife库我们看AbstractLifecycle类,如下:

public abstract class AbstractLifecycle<T> extends AtomicReference<T> implements Disposable {

    //省略部分代码
    protected final void addObserver() throws Exception {
        //Lifecycle添加监听器需要在主线程执行
        if (isMainThread() || !(scope instanceof LifecycleScope)) {
            addObserverOnMain();
        } else {
            final Object object = mObject;
            AndroidSchedulers.mainThread().scheduleDirect(() -> {
                addObserverOnMain();
                synchronized (object) {
                    isAddObserver = true;
                    object.notifyAll(); //唤醒等待的线程
                }
            });
            synchronized (object) { //加锁等待
                while (!isAddObserver) {
                    try {
                        object.wait();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
        }
    }
        //省略部分代码
} 

可以看到,RxLife对子线程做了额外的操作,在子线程通过同步锁,添加完观察者后再往下走,且RxLife同样会在事件结束或者页面销毁时移除观察者。

#/   小彩蛋   /

RxLife类里面的as系列方法,皆适用于Observable、Flowable、ParallelFlowable、Single、Maybe、Completable这6个被观察者对象,道理都一样,这里不在一一讲解。另外,在Activity/Fragment上,如果你想在某个生命周期方法中断管道,可使用as操作符的重载方法,如下:

//在Activity/Fragment上
Observable.interval(1, 1, TimeUnit.SECONDS)  //隔一秒发送一条消息
    .as(RxLife.as(this, Event.ON_STOP)) //在onStop方法中断管道
    .subscribe(aLong -> {
        Log.e("LJX", "accept=" + aLong);
    }); 

此时如果你还想在主线程回调观察者,使用asOnMain方法即可,如下:

//在Activity/Fragment上
Observable.interval(1, 1, TimeUnit.SECONDS)  //隔一秒发送一条消息
    .as(RxLife.asOnMain(this, Event.ON_STOP)) //在onStop方法中断管道,并在主线程回调观察者
    .subscribe(aLong -> {
        Log.e("LJX", "accept=" + aLong);
    });

//等同于
Observable.interval(1, 1, TimeUnit.SECONDS)  //隔一秒发送一条消息
    .observeOn(AndroidSchedulers.mainThread())
    .as(RxLife.as(this, Event.ON_STOP)) //在onStop方法中断管道,并在主线程回调观察者
    .subscribe(aLong -> {
        Log.e("LJX", "accept=" + aLong);
    });                                                       

#/   写在最后   /

在Activity/Fragment/View中,无需做任何准备工作就可以直接使用RxLife.as(this), 然而在ViewModel及任意类,需要分别继承ScopeViewModel及BaseScope类才可以使用RxLife.as(this),这多少都带有点侵入性,但这也是没有办法的办法,如果你觉得这样不能接受,RxLife允许你自行去实现Scope接口。

注:一定要使用ViewModelProviders获取ViewModel对象,如下:

//在Activity/Fragment上
MyViewModel viewModel = ViewModelProviders.of(this).get(MyViewModel.class) 

如文章中若有疏漏之处,请广大读者指正,RxLife刚出来不久,使用过程中如有遇到问题,请在github上留言。

GitHub示例:

https://github.com/liujingxing/RxLife读者福利分享