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

Posted 2022-12-15 江湖人称小白哥

在前边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