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请求框架
