阅读 814

Android RxLife 一款轻量级别的RxJava生命周期管理库(一)

简介

RxLife是一款轻量级别的RxJava生命周期管理库,代码侵入性极低,随用随取,不需要做任何准备工作,支持在Activity/Fragment 的任意生命周期方法断开管道。

原理

RxLife通过Jetpack 下的 Lifecycle 获取 Activity/Fragment 的生命周期变化,并通过Observable.lift(ObservableOperator) 操作符,注入自己实现的Observer对象(该对象能感知 Activity/Fragment的生命周期变化),从而在onSubscribe(Disposable d)方法中拿到Disposable对象,随后在相应的生命周期回调里执行Disposable.dispose()方法断开管道,这样就能将lift操作符上面的所有Disposable对象全部断开。

为什么要重复造轮子

熟悉RxJava的同学应该都知道trello/RxLifecycle 项目,它在目前的3.0.0版本中通过Lifecycle感知Activity/Fragment 的生命周期变化,并通过BehaviorSubject类及composetakeUntil操作符来实现管道的中断,这种实现原理有一点不足的是,它在管道断开后,始终会往下游发送一个onComplete事件,这对于在onComplete事件中有业务逻辑的同学来说,无疑是致命的。那为什么会这样呢?因为takeUntil操作符内部实现机制就是这样的,有兴趣的同学可以去阅读takeUntil操作符的源码,这里不展开。而RxLife就不会有这样问题,因为在原理上RxLife就与trello/RxLifecycle不同,并且RxLife还在lift操作都的基础上提供了一些额外的api,能有效的避免因RxJava内部类持有Activity/Fragment的引用,而造成的内存泄漏问题,下面开始讲解。

gradle依赖

implementation 'com.rxjava.rxlife:rxlife:1.0.4'
复制代码

源码下载

用法

Observable.timer(10, TimeUnit.SECONDS)
        //默认在onDestroy时中断管道
        .lift(RxLife.lift(this))
        .subscribe(aLong -> {
            Log.e("LJX", "accept =" + aLong);
        });
//或者
Observable.timer(10, TimeUnit.SECONDS)
        //指定在onStop时中断管道
        .lift(RxLife.lift(this,Event.ON_STOP))
        .subscribe(aLong -> {
            Log.e("LJX", "accept =" + aLong);
        });
复制代码

在Activity/Fragment 中,使用Observable的lift()操作符,方法中传入RxLife.lift(this),如果需要指定生命周期方法,额外再传一个Event对象即可。怎么样??是不是极其简单,根本不需要做任何准备工作,代码侵入性极低。

处理内存泄漏

我们来看一个案例

public void leakcanary(View view) {
    Observable.timer(100, TimeUnit.MILLISECONDS)
            .map(new MyFunction<>()) //阻塞操作
            .lift(RxLife.lift(this))
            .subscribe(new Consumer<Long>() { //这里使用匿名内部类,持有Activity的引用
                //注意这里不能使用Lambda表达式,否则leakcanary检测不到内存泄漏
                @Override
                public void accept(Long aLong) throws Exception {
                    Log.e("LJX", "accept =" + aLong);
                }
            });
}

//这里使用静态内部类,不会持有外部类的引用
static class MyFunction<T> implements Function<T, T> {

    @Override
    public T apply(T t) throws Exception {
        //当dispose时,第一次睡眠会被吵醒,接着便会进入第二次睡眠
        try {
            Thread.sleep(3000);
        } catch (Exception e) {

        }

        try {
            Thread.sleep(30000);
        } catch (Exception e) {

        }
        return t;
    }
}
复制代码

上面的代码会造成Activity无法回收,导致内存泄漏,我们用Leakcannry工具来检测一下,发现确实造成来内存泄漏,如下

在这里插入图片描述
我们已经使用RxLife库,会自动中断管道,那为什么还会造成内存泄漏呢?其实原因很简单,我们只是中断了管道,而没有中断上游对下游引用。看上面的截图就能知道,上游始终持有下游的引用,而最下游的匿名内部类Consumer又持有了Activity的引用,所以就导致了Activity无法回收。

那为什么中断管道时,不会中断上下游的引用呢?

首先有一点我们需要明确,调用Disposable.dispose()方法来断开管道,并不是真正意义上的将上游与下游断开,它只是改变了管道上各个Observer对象的一个标志位的值,我们来看一下LambdaObserver类的源码就会知道

@Override
    public void dispose() {
        DisposableHelper.dispose(this);
    }
复制代码

呃呃,只有一行代码,我们继续

public static boolean dispose(AtomicReference<Disposable> field) {
        Disposable current = field.get(); //此处得到上游的Disposable对象
        Disposable d = DISPOSED;
        if (current != d) {
            current = field.getAndSet(d); //更改自己的标志位为DISPOSED
            if (current != d) {
                if (current != null) {
                    current.dispose();//关闭上游的Disposable对象
                }
                return true;
            }
        }
        return false;
    }
复制代码

可以看到,这里只做了两件事,一是更改自己的标志位,二是调用上游的dispose()方法,其实你只要多看看,你就发现,RxJava内部大多数Observer在dispose()方法都会干这两件事。

到这,我们该如何解决这个内存泄漏问题呢?其实,RxJava早就想到了这一点,它给我们提供了一个onTerminateDetach()操作符,这个操作符会在onError(Throwable t)onComplete()dispose()这个3个时刻,断开上游对下游的引用,我们来看看源码,源码在ObservableDetach类中

@Override
public void dispose() {
    Disposable d = this.upstream;
    this.upstream = EmptyComponent.INSTANCE;//上游重新赋值
    this.downstream = EmptyComponent.asObserver();//下游重新赋值
    d.dispose();//调用上游的dispose()方法
}

@Override
public void onError(Throwable t) {
    Observer<? super T> a = downstream;
    this.upstream = EmptyComponent.INSTANCE;//上游重新赋值
    this.downstream = EmptyComponent.asObserver();//下游重新赋值
    a.onError(t); //调用下游的onError方法
}

@Override
public void onComplete() {
    Observer<? super T> a = downstream;
    this.upstream = EmptyComponent.INSTANCE;//上游重新赋值
    this.downstream = EmptyComponent.asObserver();//下游重新赋值
    a.onComplete();//调用下游的onComplete方法
}

复制代码

到这,我们就知道该怎么做了,下面这样写就安全了

Observable.timer(100, TimeUnit.MILLISECONDS)
        .map(new MyFunction<>())//阻塞操作
        .onTerminateDetach() //管道断开时,中断上游对下游的引用
        .lift(RxLife.lift(this)) //默认在onDestroy时断开管道
        .subscribe(aLong -> {
            Log.e("LJX", "accept =" + aLong);
        });
复制代码

可是,每次都要这样写吗?有没有更简单的,有,RxLife提供了RxLife.compose(LifecycleOwner)方法,内部就是将onTerminateDetachlift这两个操作符整合在了一起,接下来,看看如何使用

Observable.timer(100, TimeUnit.MILLISECONDS)
        .map(new MyFunction<>())//阻塞操作
         //注意这里使用compose操作符
        .compose(RxLife.compose(this))//默认在onDestroy时中断管道,并中断下下游之间的引用
        .subscribe(aLong -> {
            Log.e("LJX", "accept =" + aLong);
        });
复制代码

如果需要指定生命周期的方法,也可以

Observable.timer(100, TimeUnit.MILLISECONDS)
        .map(new MyFunction<>())//阻塞操作
         //注意这里使用compose操作符
        .compose(RxLife.compose(this, Event.ON_STOP))//指定在onStop时断开管道
        .subscribe(aLong -> {
            Log.e("LJX", "accept =" + aLong);
        });
    }
复制代码

大多数情况下,我们希望观察者能主线程进行回调,也许你会这样写

Observable.timer(100, TimeUnit.MILLISECONDS)
        .map(new MyFunction<>())//阻塞操作
        .observeOn(AndroidSchedulers.mainThread()) //在主线程回调
        .compose(RxLife.compose(this, Event.ON_STOP))//指定在onStop回调时中断管道,并中断上下游引用
        .subscribe(aLong -> {
            Log.e("LJX", "accept =" + aLong);
        });
复制代码

如果你是用RxLife的话,就可以这样写,使用RxLife.composeOnMain方法

Observable.timer(100, TimeUnit.MILLISECONDS)
        .map(new MyFunction<>())//阻塞操作
        //在主线程进程回调,在onStop回调时中断管道,并中断上下游引用
        .compose(RxLife.composeOnMain(this, Event.ON_STOP))
        .subscribe(aLong -> {
            Log.e("LJX", "accept =" + aLong);
        });
复制代码

RxLife类就只有6个静态方法,如下

shshs

注意,前方高能预警!!!!!!!

结合RxLife使用Observable的liftcompose操作符时,下游除了subscribe操作符外最好不要有其它的操作符,前面讲过,当调用Disposable.dispose()时,它会往上一层一层的调用上游的dispose()方法,如果下游有Disposable对象,是调用不到的,如果此时下游有自己的事件需要发送,那么就无法拦截了。 如:

Observable.just(1)
        .compose(RxLife.compose(this))
        .flatMap((Function<Integer, ObservableSource<Long>>) integer -> {
            //每隔一秒发送一个数据,共10个
            return Observable.intervalRange(0, 10, 0, 1, TimeUnit.SECONDS);
        })
        .subscribe(aLong -> {
            Log.e("LJX", "accept =" + aLong);
        });
复制代码

这样,即使Activity关闭了,观察者每隔一秒后,依然能收到来自上游的事件,因为compose无法切断下游的管道,我们改一下上面的代码

Observable.just(1)
        .flatMap((Function<Integer, ObservableSource<Long>>) integer -> {
            //每隔一秒发送一个数据,共10个
            return Observable.intervalRange(0, 10, 0, 1, TimeUnit.SECONDS);
        })
        .compose(RxLife.compose(this))
        .subscribe(aLong -> {
            Log.e("LJX", "accept =" + aLong);
        });
复制代码

这样ok了,其实这不是RxLife的问题,使用鼎鼎大名的trello/RxLifecycle库也是一样的,因为RxJava的设计就是如此,上游拿不到下游的Disposable对象,所以,我们在使用RxLife时,一定要注意在lift或者compose操作符的下游,除了subscribe操作符外最好不要有其它的操作符,这一点一定需要注意。

RxLife最新版本已经使用as操作符规避这个问题,详情查看Android RxLife 一款轻量级别的RxJava生命周期管理库(二)

小彩蛋

RxLife类里面的life、compose系列方法,皆适用于Flowable、Observable、Single、Maybe、Completable这5个被观察者对象,道理都一样,这里不在一一讲解。

结尾

Ok,RxLife的使用基本就介绍完了,到这我们会发现,使用RxLife库,我们只需要关注一个类即可,那即是RxLife类,api简单功能却强大。敢兴趣的同学,可以去阅读RxLife源码,有疑问,请留言,我会在第一时间作答。

扩展

RxLife结合RxHttp发送请求,简直不要太爽。

RxHttp详情请点击RxHttp 一条链发送请求,新一代Http请求神器(一)

关注下面的标签,发现更多相似文章
评论