打造RxJava生命周期管理框架RxLife

在前边RxJava实战技巧大全一文中，我们介绍了RxJava使用过程中常见的应用场景和可能遇到的问题，其中我们谈到利用RxLifeCycle来管理RxJava的生命周期，避免内存泄漏问题，今天自己动手打生命周期管理框RxLife来加深对RxJava的认识。

详解Subject

什么是Subject

在RxJava当中，有四个对象值得我们关注：Observable，Subject，Observer，Subscriber，它们之间的关系如下：

对于Observable，Observer，Subscriber我们比较熟悉，故不做说明，重点来看Subject。

通过上面的图我们可以看出Subject继承自Observable，也就意味着Subject可以作为被观察者，另外，它又实现了Observer接口，这意味着它也可以作为观察者。不难看出，Subject既能作为Observer订阅Observable,又能作为Observable被其他Observer订阅。总之，Subject承担了这么一种角色：对上作为观察者，对下作为被观察者。

和Observable必须有订阅者才能发射数据不一样，无论Subject是否有订阅者，它都可以发射数据。这有点类似广播电台，不会因为我们关闭收音机就停止广播，在收听的人自然收听的到，没收听的人也无关紧要。

常见的Subject

从上面的uml中我们看出，RxJava为我们提供了四种常用的Subject，即

AsyncSubject,BehabviorSubject,PublishSubject,ReplaySubject，下面我们对这四者进行说明：

AsyncSubject

AsyncSubject会缓存最后一个数据并在调用onCompleted()时将该数据发送给订阅者，原理如下：

在该过程中，一旦发生任何异常都不会发送数据到订阅者，而是发送给订阅者一个异常通知，即订阅者只能接受到一个异常的通知，如下：

举例来说明AsyncSubject的用法：

asyncSubject.onNext("1");
asyncSubject.onNect("2");
asyncSubject.onCompleted();//必须调用才会开始发送数据

以上代码执行后，订阅者接受到的数据是2.

BehaviorSubject

当BehaviorSubject被订阅后，它首先会发送原始Observable最近发射的数据，如果最近没有，会发射一个默认值，接下继续发射原始Observable的数据，如下图：

如果原始的Observable因为发生了错误而终止，那么BehaviorSubject在发送一个错误通知后不再发射数据，如下：

我们举例来说明BehabviorSubject的用法：

behaviorSubject.onNext("1");
behaviorSubject.onNect("2");
behaviorSubject.onNext("3");
behaviorSubject.subscribe(new Action<String>(){
    @Override
    public void call(String s){
        System.out.println(“result:”+s);
    }
});
behaviorSubject.onNext("4");

输出结果是3,4.

PublishSubject

默认情况下，RxJava中的Observable一旦被订阅就开始发送事件，这和我们传统的观察者模式有所区别。而PublishSuject的行为则类似传统的观察这模式，观察者可以先订阅被观察者，然后在某个时刻手动调用方法来发射数据（订阅之后的数据）到所有的观察者。如下图：

如果原始的Observable因为发生了错误而终止，那么PublishSubject在发送一个错误通知后不再发射数据，如下：

举例来说明PublishSubject的用法：

publishSubject.onNext("1");
publishSubject.onNect("2");
publishSubject.onNext("3");//订阅之前不会被发送

publishSubject.subscribe(new Action<String>(){
    @Override
    public void call(String s){
        System.out.println(“result:”+s);
    }
});
publishSubject.onNect("4");
publishSubject.onNect("5");

1，2，3是在订阅之前的数据，不会被发射，最终输出结果是4，5。

ReplaySubject

ReplaySubject会缓存所有已经发射的数据，当一个新的订阅关系产生时，ReplaySuject会将所有数据都发送给他。另外，ReplaySubject支持设置缓存数据和缓存时间。如下图：

举例来说明ReplaySubject的用法：

replaySubject.onNext("1");
replaySubject.onNect("2");
replaySubject.onNext("3");

replaySubject.subscribe(new Action<String>(){
    @Override
    public void call(String s){
        System.out.println(“result:”+s);
    }
});
replaySubject.onNect("4");

默认情况下ReplaySubject会缓存所有的数据，因此最终数据的结果如下：

result:1
result:2
result:3
result:4

小结

回顾上面所谈的，不难看出不同的Subject最大的区别在于发送数据的行为不同，简单概括如下：

Subject 发射行为
AsyncSubject 不论订阅发生在什么时候，只会发射最后一个数据
BehaviorSubject 发送订阅之前一个数据和订阅之后的全部数据
ReplaySubject 不论订阅发生在什么时候，都发射全部数据
PublishSubject 发送订阅之后全部数据

关于Subject更详细的使用方法请直接查阅api doc.

实现生命周期管理框架（RxLife）

在了解Subject之后就可以开始考虑如何实现一个生命周期管理框架。每当Activity或者Fragment的生命周期发生变化时我们都希望产生一个对应的事件来通知当前所有的订阅者，这样我们就可以根据对应的事件去确定是否取消订阅关系了。

从上面的描述中，我们有两个问题要解决：

  1. 如何监听Activity或Fragmeng生命周期变化并将其发送出去。
  2. 原有的观察者如何接受生命周期，并在某生命周期下中断原有的Observable。

通过以上两个问题，我们知道我们需要一个既能够发射生命周期，又能接受生命周期的观察者，因此不难想到这里需要Subject。生命周期是连续产生的，无论是否有订阅者，我们只关注最最近的生命周期，因此我们选择使用BehaviorSubject。

现在我们来考虑如何监听Activity或Fragment的生命周期，并利用BehaviorSubject发射生命周期。这里我们以Activity为例进行说明。

生命周期事件监听

定义生命周期事件

我们根据Activity的生命周期，定义相应的事件。

public enum  ActivityEvent {
    CREATE,
    RESUME,
    START,
    PAUSE,
    STOP,
    DESTORY
}

监听生命周期

为了能在Activitiy生命周期变化时发送相应的事件，我们定义了RxAppcompatActivity，该类继承了AppCompatActivity并重写器生命周期方法：在不同方法中发射事件到BehaviorSubject中。这就好像我们的BehaviorSubject对象在不断的观察Activity生命周期的变化。当然，由于Subject的特性，BehaviorSubject也具备了将这些事件发射出去的能力。

public class RxAppCompatActivity extends AppCompatActivity {
    protected final BehaviorSubject<ActivityEvent> lifeSubject = BehaviorSubject.create();

    @Override
    protected void onCreate(@Nullable Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        lifeSubject.onNext(ActivityEvent.CREATE);
    }

    @Override
    protected void onResume() {
        super.onResume();
        lifeSubject.onNext(ActivityEvent.RESUME);
    }

    @Override
    protected void onStart() {
        super.onStart();
        lifeSubject.onNext(ActivityEvent.START);
    }

    @Override
    protected void onPause() {
        super.onPause();
        lifeSubject.onNext(ActivityEvent.PAUSE);
    }

    @Override
    protected void onStop() {
        super.onStop();
        lifeSubject.onNext(ActivityEvent.STOP);
    }

    @Override
    protected void onDestroy() {
        super.onDestroy();
        lifeSubject.onNext(ActivityEvent.DESTORY);
    }
}

Observable自动停止发射数据

到现在我们已经利用Subject来监视生命周期的变化，那又如何让原有的Observable（比如网络请求的Observable）来监视Subject发射的数据呢，并根据Subject的状态自动停止原始数据的发射？换言之就是一个Observable如何在发射数据的同时监视另一个Observable？

TakeUtil操作符

令人高兴的是，RxJava中提供的TakeUntil操作符来实现上述需求。TakeUntil订阅原始的Observable并发射数据，此外它还监视你提供的第二个Observable。当第二个Observable发射了一项数据或者发射一项终止的通知时（onError通知或者onCompleted通知），TakeUntil返回的Observable会停止发射原始的Observable，如下图所示：

我们用一个简单的例子来展示TakeUntil操作符的使用：

 Observable.interval(2, TimeUnit.SECONDS).subscribe(new Action1<Long>() {
            @Override
            public void call(Long num) {
                Log.d("MainActivity", "num:" + num);
            }
        });

上面的代码每隔2s进行输出，现在我们希望5s后自动停止输出,就可以这样做：

   Observable.interval(2, TimeUnit.SECONDS).takeUntil(Observable.timer(5,TimeUnit.SECONDS)).subscribe(new Action1<Long>() {
            @Override
            public void call(Long num) {
                Log.d("MainActivity", "num:" + num);
            }
        });

为了让以上代码更通用，我们利用compose操作符进行改写（对compose不熟悉的童鞋自行查阅资料）：

private void startIntervalTask1() {
        Observable.interval(2, TimeUnit.SECONDS).compose(bindUntilDelay(5)).subscribe(new Action1<Long>() {
            @Override
            public void call(Long num) {
                Log.d("MainActivity", "num:" + num);
            }
        });

    }

    @NonNull
    private Observable.Transformer<Long, Long> bindUntilDelay(final int delaySecond) {
        return new Observable.Transformer<Long, Long>() {
            @Override
            public Observable<Long> call(Observable<Long> longObservable) {
                return longObservable.takeUntil(timer(delaySecond,TimeUnit.SECONDS));
            }
        };
    }

回到正题，现在我们已经有了可以发射生命周期事件的BehaviorSubject，再结合TakeUntil不就可以实现在指定生命周期发生时自动停止原有的Observable了吗？

结合BehaviorSubject与TakeUntil

有了上面的知识做铺垫，实现生命周期管理框架也就显得轻而易举了。

为了方便使用，我们在RxAppcompatActivity中提供了bindUntilEvent(ActivityEvent nindEvent)方法：

public class RxAppCompatActivity extends AppCompatActivity {
    protected final BehaviorSubject<ActivityEvent> lifeSubject = BehaviorSubject.create();

    public <T> Observable.Transformer<T, T> bindUntilEvent(final ActivityEvent bindEvent) {
        //被监视的Observable
        final Observable<ActivityEvent> observable = lifeSubject.takeFirst(new Func1<ActivityEvent, Boolean>() {
            @Override
            public Boolean call(ActivityEvent event) {
                return event.equals(bindEvent);
            }
        });

        return new Observable.Transformer<T, T>() {
            @Override
            public Observable<T> call(Observable<T> sourceOb) {
                return sourceOb.takeUntil(observable);
            }
        };
    }

    @Override
    protected void onCreate(@Nullable Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        lifeSubject.onNext(ActivityEvent.CREATE);
    }

    @Override
    protected void onResume() {
        super.onResume();
        lifeSubject.onNext(ActivityEvent.RESUME);
    }

    @Override
    protected void onStart() {
        super.onStart();
        lifeSubject.onNext(ActivityEvent.START);
    }

    @Override
    protected void onPause() {
        super.onPause();
        lifeSubject.onNext(ActivityEvent.PAUSE);
    }

    @Override
    protected void onStop() {
        super.onStop();
        lifeSubject.onNext(ActivityEvent.STOP);
    }

    @Override
    protected void onDestroy() {
        super.onDestroy();
        lifeSubject.onNext(ActivityEvent.DESTORY);
    }
}

接下来，我们用同样的方式来处理Fragment或者其他组件即可。

具体使用

新建的Activity需要继承我们的RxAppcompatActivity，新建的Fragment则继承我们的RxFragment，就是这么简单。

我们同样还是以师父说为例，由于我们的方法基本和RxLifeCycle保持一致，因此只要简单的改动就可以让RxLife工作起来，现在就可以用RxLife来代替RxLifeCycle。

仍然做个简单的示例：

ApiFactory.getWXApi().getWXHot(AppConstant.KEY_WX, getPageSize(), mCurrentPage + 1).compose(this.bindUntilEvent(FragmentEvent.STOP))
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(mSubscriber);

总结

通过自行实现一个RxJava生命周期管理框架（RxLife）加深le我们对RxJava中Subject的理解。另外，Subject的应用非常广泛，在下篇文章中，我们将会进一步深入，利用Subject来打造自己的事件通信总线RxBus。

github：https://github.com/closedevice/FastApp

时间： 2024-08-01 12:54:57

打造RxJava生命周期管理框架RxLife的相关文章

助力绵阳市商业银行，打造高效项目生命周期管理平台

金融市场捷报连连,近日,TechExcel公司再次凭借产品和服务实力直签下绵阳市商业银行,打造项目全生命周期管理平台.DevSuite研发与项目管理平台为绵阳市商业银行员工提供了一套必不可少的信息化工具,大大提升研发管理效率. 绵阳市商业银行(绵阳市商业银行股份有限公司)是国有控股地方商业银行,成立于2000年9月,是中国(绵阳)科技城唯一一家国有控股法人金融机构.成立至今,绵阳市商业银行秉承"服务地方.服务中小"的企业宗旨,全力打造一流中小企业金融服务机构,不断开创科学发展新局面.2

RxJava 和 RxAndroid 三（生命周期控制和内存优化）

RxJava 和 RxAndroid 三(生命周期控制和内存优化) rxjava rxandroid 前言:对Rxjava.Rxandroid不了解的同学可以先看看 RxJava 和 RxAndroid RxJava 和 RxAndroidRxJava 和 RxAndroid 二(操作符的使用) RxJava使我们很方便的使用链式编程,代码看起来既简洁又优雅.但是RxJava使用起来也是有副作用的,使用的越来越多的订阅,内存开销也会变得很大,稍不留神就会出现内存溢出的情况,这篇文章就是介绍Rxj

Unity3D脚本(MonoBehaviour)生命周期分析

欢迎来到unity学习.unity培训.unity企业培训教育专区,这里有很多U3D资源.U3D培训视频.U3D教程.U3D常见问题.U3D项目源码,我们致力于打造业内unity3d培训.学习第一品牌. 场景中有2个物体:A,B 每个物体上绑定2个脚本:A,B 初始化log: Object : A , Script : B , Message : Awake Object : A , Script : B , Message : OnEnable Object : A , Script : A

连载《一个程序猿的生命周期》-《发展篇》- 6.2016年发展元年，本职工作和个人事业均有突破和起色

摘要:社会的一种病态,不停的在打鸡血,社会带来N座大山却又在无情让你绝望.要么自求多福,要么寻找突破:要么等死,要么在改变中死,反正都是死,需要自己做出选择. 很长时间没有写文章了,但是自从写<一个程序猿的生命周期>以后感觉写文章不再是一件困难的事,因为习惯和思考融入了生活和工作,无时无刻的观察.思考.反思.有人会问这样的生活不感觉累吗?说不累是骗人的,但是也能感受到其中的快乐,大大的抵消了所谓的"累". 这一年以来的确很忙,能感受到强烈的时间压迫感,似乎有很多需要去作的&

NET：关于ChenkBoxList生命周期

前几天写了个弹出层,显示多选功能,但一直出错, 出错原因及代码如链接:http://bbs.csdn.net/topics/390820888 搞定之后,仔细回想下,觉得自己没有意识到控件的生命周期问题. 弹出层里面放的checkboxlist的话,那么每选中一个则页面会回发到服务器,然后服务器会解析页面模版并返回html给客户端,客户端重新解析html并渲染,重新生成新页面,且已保留页面状态值. 在这个过程中,回发时,会重新载入PageLoad函数,此刻,在这里写一个是否回发函数,肯定不执行回

【深入ASP.NET原理系列】--ASP.NET页面生命周期

前言 ASP.NET页面运行时候,页面将经历一个生命周期,在生命周期中将执行一系列的处理步骤.包括初始化.实例化控件.还原和维护状态.运行时间处理程序代码以及进行呈现.熟悉页面生命周期非常重要,这样我们才能在生命周期的合适阶段编写代码.如果我们能在写代码的时候想着我们现在是在做生命周期的哪一步那将是非常好的. 你可能会说我不清楚还不是一样写代码,反正每次都在Page_load里面写代码 然后页面就出来了我管那么多干什么.所谓知其然如果能知其所以然岂不是更吊?我个人认为做ASP.NET B/S开发

管道及页面生命周期

客户端请求发送到服务器进入管道执行页面生命周期: 客户发送请求到服务器,服务器通过扩展名扫描映射表,交给对应的扩展程序(aspnet_isapi.dll),扩展程序把请求传入framwork,即调用ISAPIRuntime pr方法,调用HttpRuntime pr方法,会创建HttpWorkerRequest,将请求封装到HttpContext,通过HttpApplicationFactory创建HttpApplication负责处理整个请求--调用pr方法,将HttpContext传入app

WebForm页面生命周期及asp.net运行机制

1.先上几张原理图着重理解: 现在针对第四副图原理进行解析: 流程: 1.浏览器发送请求 2.服务器软件(IIS)接收,它最终的目的就是为了向客户输出它请求的动态页面生成的html代码. 3.服务器不会处理类和动态页面,所以找扩展程序 4.交给FrameWork,它其中有个类HttpRuntime,其中有个ProcessRequest()方法 5.ProcessRequest()创建了HttpWorkerRequest对象,把请求报文都封装在其中(HttpWorkerRequest中有个指针,它

逸管家：把握不同行业生命周期，选择正确商业模式

</P><P> 原标题:把握不同行业生命周期,选择正确商业模式</P><P> 现今中小企业已进入理性商业模式选择期,大量中小企业在拿到天使投资后进入融资瓶颈期.与此同时,"共享"一词在各地出现,但缺少创业项目运营服务管理能力,能提供的仅仅是基本的公司注册管理和场地服务,无法解决中小企业的软服务提供问题.然而,对于中小企业来说,真正要走向成功必须考虑好三个问题:</P><P> 第一.选择什么样的商业模式;</P