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类及compose、takeUntil操作符来实现管道的中断，这种实现原理有一点不足的是，它在管道断开后，始终会往下游发送一个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)方法，内部就是将onTerminateDetach、lift这两个操作符整合在了一起，接下来，看看如何使用


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的lift、compose操作符时，下游除了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结合HttpSender发送请求，简直不要太爽。

HttpSender详情请点击HttpSender OkHttp+RxJava超好用、功能超级强大的Http请求框架