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的引用,而造成的内存泄漏问题,下面开始讲解。
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
操作符外最好不要有其它的操作符,这一点一定需要注意。
RxLife类里面的6个静态方法,皆适用于Flowable、Observable、Single、Maybe、Completable这5个被观察者对象,道理都一样,这里不在一一讲解。
Ok,RxLife的使用基本就介绍完了,到这我们会发现,使用RxLife库,我们只需要关注一个类即可,那即是RxLife类,api简单功能却强大。敢兴趣的同学,可以去阅读 RxLife
的 源码 ,有疑问,请留言,我会在第一时间作答。
RxLife 结合 HttpSender 发送请求,简直不要太爽。
HttpSender详情请点击 HttpSender OkHttp+RxJava超好用、功能超级强大的Http请求框架