转载

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.3'
复制代码

源码下载

用法

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个静态方法，如下

注意，前方高能预警！！！！！！！

结合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 操作符外最好不要有其它的操作符，这一点一定需要注意。