如何开发一个简单好用的RxBus

作者：拉丁吴 https://www.jianshu.com/p/b5339f7bdfb3 已获作者授权转载

在Android开发中，事件总线的库往往是开发必备的利器之一，我经历的几个开发项目，都无一例外的引用了事件总线的库，因为它能帮助我们非常简单的实现组件之间的通信工作，极大的提高开发效率。

市面上EventBus，RxBus都是比较成熟的库，为什么还是考虑自己开发一个呢？

主要还是用着不太顺手，首先，Rxbus不支持粘性事件，这也就意味着Activity/Fragment之间的数据传递，还是需要写很多Intent之类的代码，降低了开发效率。其次，Rxbus,EventBus都需要手动注册和注销，稍显麻烦。当然，最重要的因素是，通过RxJava开发一个RxBus也很方便。

因此在尝试开发自己的RxBus之前，重点考虑下面两点：

支持粘性事件 这是非常重要的，因为有了粘性事件，我们可以解决activity/Fragment之间的消息传递，附带的好处是，进程内的数据传递，可以打破Intent的大小限制。

简单易用 这个是一个很重要的原则，参考rxbus,我们依然需要处理register(),unregister()方法，这就不是很友好了。一个成熟的rxbus应该能够学会自己注册和注销，作为使用者，我们只需要关心发送事件和接收事件。

如何支持粘性事件

RxJava天然的支持事件分发传递，比如，普通事件的传递，发送端我们可以直接使用PublishSubject,接收端则是普通的Observable即可，但是要支持粘性事件，我们需要考虑的东西就没这么简单了，首先使用什么样的Subject能达成这样的效果，是否有功能隐患或者性能隐患？

先来看看如何实现粘性事件的功能，我们熟知的Subject有四种：AsyncSubject，BehaviorSubject，PublishSubject，ReplaySubject，我们先一一解释下这些东西

• AsyncSubject：只在原始Observable完成后，发射来自原始Observable的最后一个值

• BehaviorSubject：发射原始Observable最近发射的数据

• PublishSubject：会把在订阅发生的时间点之后来自原始Observable的数据发射给观察者

• ReplaySubject： 会发射所有来自原始Observable的数据给观察者

从上面的介绍可以看出，AsyncSubject显然不合适，PublishSubject看起来也不太合适，因为它不会发送订阅之前的消息，ReplaySubject和BehaviorSubject都能发送订阅之前的消息。

ReplaySubject最大的问题就是它会把发送的历史消息都存起来，但是我们其实并不需要存储所有的事件，如果事件太多会带来不必要的内存负担，虽然ReplaySubject能提供方法设置内部最大存储量来控制存储大小，但是无法细粒度的定点清除事件，因此，我们先把它作为一个性能较差备选的方案。

BehaviorSubject它只会存储最近的一个事件，这样不会有内存隐患，但是这个特性本身也会存在隐患。比如，在发送事件A和接收事件A之间的某个时间点，如果又发送了事件B 那么，事件A就会被抛弃。接受者就永远无法收到事件A了。这一点，从下图中也很容易看得出来。

这个问题不容易被发现，开发人员能够意识到这个问题还可以避免，但是如果多人协作，项目越来越复杂的情况下，我们就很难保证不会出现这样的问题了。因此，BehaviorSubject也不是一个好的选择。

其实，以上四个Subject都不是最好的选择，最终还是决定自己缓存事件，并在合适的时机清除历。

添加事件的时机是当我们需要post(event)的时候，就把事件添加进来，那么何时清除事件呢？是消费完成之后就清除？显然是不太合理的，参考Intent，在Activity中，可以多次获取Intent，之后activity被销毁了，intent才会被销毁，因此，我们清除事件在取消订阅的时候，也就是组件被销毁的时候。保证了我们可以多次多地获取同样额数据，

自动注册注销

在新的事件总线库中，只有post(event)和receive(event),至于注册和注销我们基本不需要处理。除了我们所关心的，没有任何多余的工作。

框架会帮你自动注册和注销。注册发生在准备接收数据的时候，即调用receive(Message)时，而注销的时机就显然是当前组件被销毁的时候，因此，我们通过构造一个无界面的Fragment添加到当前的Activity中来实现监听当前组件的生命周期。

SimpleRxBus

SimpleRxBus就是按照上述想法来开发的一个事件总线库，地址：https://github.com/ladingwu/SimpleRxBusApplication，以下是使用简介：

集成

implementation 'com.ladingwu.library:SimpleRxBus:0.1'
// 需要v7的support包，如果项目中已经有了，则不用添加
implementation 'com.android.support:appcompat-v7:28.0.0'  

发送普通事件

RxBusUtils.post("filter_message",mesage);
接收事件(自动取消订阅)

// in Fragment or FragmentActivity
 RxBusUtils.receive(this,"filter_message", new RxBusReceiver<Object>() {
            @Override
            public void receive(Object message) {
                // handle this
            }
        });

发送粘性事件

RxBusUtils.postSticky("filter_sticky_message",message);

接收粘性事件(自动取消订阅)

// in Fragment or FragmentActivity
RxBusUtils.receiveSticky(this,"filter_sticky_message", new RxBusReceiver<Object>() {
            @Override
            public void receive(Object message) {
                    // handle this
            }
        });

特殊情况

//如果无法拿到Fragment/FragmentActivity的实例，则接收事件的时候，需要自行处理取消注册的工作
    Disposable disposable = RxBusUtils.receive("filter", new RxBusReceiver<Object>() {
        @Override
        public void receive(Object data) {
            // handle this
        }
    });

    // 在合适的时机取消注册
    if (disposable != null && !disposable.isDisposed()) {
        disposable.dispose();
    }

— — — END — — —