Android消息总线的演进之路：用LiveDataBus替代RxBus、EventBus

淡淡的烟草味﹌ 2024-02-19 21:44 60阅读 0赞

![640?wx\_fmt=png][640_wx_fmt_png]

**总第269篇**

**2018年 第61篇**

**背景**

对于Android系统来说，消息传递是最基本的组件，每一个App内的不同页面，不同组件都在进行消息传递。消息传递既可以用于Android四大组件之间的通信，也可用于异步线程和主线程之间的通信。对于Android开发者来说，经常使用的消息传递方式有很多种，从最早使用的Handler、BroadcastReceiver、接口回调，到近几年流行的通信总线类框架EventBus、RxBus。Android消息传递框架，总在不断的演进之中。

**从EventBus说起**

EventBus是一个Android事件发布/订阅框架，通过解耦发布者和订阅者简化Android事件传递。EventBus可以代替Android传统的Intent、Handler、Broadcast或接口回调，在Fragment、Activity、Service线程之间传递数据，执行方法。

EventBus最大的特点就是简洁、解耦。在没有EventBus之前我们通常用广播来实现监听，或者自定义接口函数回调，有的场景我们也可以直接用Intent携带简单数据，或者在线程之间通过Handler处理消息传递。但无论是广播还是Handler机制远远不能满足我们高效的开发。EventBus简化了应用程序内各组件间、组件与后台线程间的通信。EventBus一经推出，便受到广大开发者的推崇。

现在看来，EventBus给Android开发者世界带来了一种新的框架和思想，就是消息的发布和订阅。这种思想在其后很多框架中都得到了应用。

![640?wx\_fmt=png][640_wx_fmt_png 1]

图片摘自EventBus GitHub主页

**发布/订阅模式**

订阅发布模式定义了一种“一对多”的依赖关系，让多个订阅者对象同时监听某一个主题对象。这个主题对象在自身状态变化时，会通知所有订阅者对象，使它们能够自动更新自己的状态。

![640?wx\_fmt=png][640_wx_fmt_png 2]

**RxBus的出现**

RxBus不是一个库，而是一个文件，实现只有短短30行代码。RxBus本身不需要过多分析，它的强大完全来自于它基于的RxJava技术。响应式编程（Reactive Programming）技术这几年特别火，RxJava是它在Java上的实作。RxJava天生就是发布/订阅模式，而且很容易处理线程切换。所以，RxBus凭借区区30行代码，就敢挑战EventBus“江湖老大”的地位。

**RxBus原理**

在RxJava中有个Subject类，它继承Observable类，同时实现了Observer接口，因此Subject可以同时担当订阅者和被订阅者的角色，我们使用Subject的子类PublishSubject来创建一个Subject对象（PublishSubject只有被订阅后才会把接收到的事件立刻发送给订阅者），在需要接收事件的地方，订阅该Subject对象，之后如果Subject对象接收到事件，则会发射给该订阅者，此时Subject对象充当被订阅者的角色。

完成了订阅，在需要发送事件的地方将事件发送给之前被订阅的Subject对象，则此时Subject对象作为订阅者接收事件，然后会立刻将事件转发给订阅该Subject对象的订阅者，以便订阅者处理相应事件，到这里就完成了事件的发送与处理。

最后就是取消订阅的操作了，RxJava中，订阅操作会返回一个Subscription对象，以便在合适的时机取消订阅，防止内存泄漏，如果一个类产生多个Subscription对象，我们可以用一个CompositeSubscription存储起来，以进行批量的取消订阅。

**RxBus有很多实现，如：**

AndroidKnife/RxBus（[https://github.com/AndroidKnife/RxBus][https_github.com_AndroidKnife_RxBus]）  
Blankj/RxBus（[https://github.com/Blankj/RxBus][https_github.com_Blankj_RxBus]）

其实正如前面所说的，RxBus的原理是如此简单，我们自己都可以写出一个RxBus的实现：

**基于RxJava1的RxBus实现：**

public final class RxBus {    private final Subject<Object, Object> bus;    private RxBus() {        bus = new SerializedSubject<>(PublishSubject.create());    }    private static class SingletonHolder {        private static final RxBus defaultRxBus = new RxBus();    }    public static RxBus getInstance() {        return SingletonHolder.defaultRxBus;    }    /*     * 发送     */    public void post(Object o) {        bus.onNext(o);    }    /*     * 是否有Observable订阅     */    public boolean hasObservable() {        return bus.hasObservers();    }    /*     * 转换为特定类型的Obserbale     */    public <T> Observable<T> toObservable(Class<T> type) {        return bus.ofType(type);    }}

**基于RxJava2的RxBus实现：**

public final class RxBus2 {    private final Subject<Object> bus;    private RxBus2() {        // toSerialized method made bus thread safe        bus = PublishSubject.create().toSerialized();    }    public static RxBus2 getInstance() {        return Holder.BUS;    }    private static class Holder {        private static final RxBus2 BUS = new RxBus2();    }    public void post(Object obj) {        bus.onNext(obj);    }    public <T> Observable<T> toObservable(Class<T> tClass) {        return bus.ofType(tClass);    }    public Observable<Object> toObservable() {        return bus;    }    public boolean hasObservers() {        return bus.hasObservers();    }}

**引入LiveDataBus的想法**

**从LiveData谈起**

LiveData是Android Architecture Components提出的框架。LiveData是一个可以被观察的数据持有类，它可以感知并遵循Activity、Fragment或Service等组件的生命周期。正是由于LiveData对组件生命周期可感知特点，因此可以做到仅在组件处于生命周期的激活状态时才更新UI数据。

LiveData需要一个观察者对象，一般是Observer类的具体实现。当观察者的生命周期处于STARTED或RESUMED状态时，LiveData会通知观察者数据变化；在观察者处于其他状态时，即使LiveData的数据变化了，也不会通知。

**LiveData的优点**

**谈一谈Android Architecture Components**

Android Architecture Components的核心是Lifecycle、LiveData、ViewModel 以及 Room，通过它可以非常优雅的让数据与界面进行交互，并做一些持久化的操作，高度解耦，自动管理生命周期，而且不用担心内存泄漏的问题。

**Android Architecture Components的特点**

**重点：为什么使用LiveData构建数据通信总线LiveDataBus**

**使用LiveData的理由**

**为什么要用LiveDataBus替代EventBus和RxBus**

**LiveDataBus的设计和架构**

**LiveDataBus的组成**

#### **LiveDataBus原理图** ####

![640?wx\_fmt=png][640_wx_fmt_png 3]

**LiveDataBus的实现**

**第一个实现：**

public final class LiveDataBus {    private final Map<String, MutableLiveData<Object>> bus;    private LiveDataBus() {        bus = new HashMap<>();    }    private static class SingletonHolder {        private static final LiveDataBus DATA_BUS = new LiveDataBus();    }    public static LiveDataBus get() {        return SingletonHolder.DATA_BUS;    }    public <T> MutableLiveData<T> getChannel(String target, Class<T> type) {        if (!bus.containsKey(target)) {            bus.put(target, new MutableLiveData<>());        }        return (MutableLiveData<T>) bus.get(target);    }    public MutableLiveData<Object> getChannel(String target) {        return getChannel(target, Object.class);    }}

短短二十行代码，就实现了一个通信总线的全部功能，并且还具有生命周期感知功能，并且使用起来也及其简单：

**注册订阅：**

LiveDataBus.get().getChannel("key_test", Boolean.class)        .observe(this, new Observer<Boolean>() {            @Override            public void onChanged(@Nullable Boolean aBoolean) {            }        });

**发送消息：**

LiveDataBus.get().getChannel("key_test").setValue(true);

我们发送了一个名为"key\_test"，值为true的事件。  
这个时候订阅者就会收到消息，并作相应的处理，非常简单。

**问题出现**

对于LiveDataBus的第一版实现，我们发现，在使用这个LiveDataBus的过程中，订阅者会收到订阅之前发布的消息。对于一个消息总线来说，这是不可接受的。无论EventBus或者RxBus，订阅方都不会收到订阅之前发出的消息。对于一个消息总线，LiveDataBus必须要解决这个问题。

**问题分析**

怎么解决这个问题呢？先分析下原因：

当LifeCircleOwner的状态发生变化的时候，会调用LiveData.ObserverWrapper的activeStateChanged函数，如果这个时候ObserverWrapper的状态是active，就会调用LiveData的dispatchingValue。

![640?wx\_fmt=png][640_wx_fmt_png 4]

在LiveData的dispatchingValue中，又会调用LiveData的considerNotify方法。

![640?wx\_fmt=png][640_wx_fmt_png 5]

在LiveData的considerNotify方法中，红框中的逻辑是关键，如果ObserverWrapper的mLastVersion小于LiveData的mVersion，就会去回调mObserver的onChanged方法。而每个新的订阅者，其version都是-1，LiveData一旦设置过其version是大于-1的（每次LiveData设置值都会使其version加1），这样就会导致LiveDataBus每注册一个新的订阅者，这个订阅者立刻会收到一个回调，即使这个设置的动作发生在订阅之前。

![640?wx\_fmt=png][640_wx_fmt_png 6]

**问题原因总结**

对于这个问题，总结一下发生的核心原因。对于LiveData，其初始的version是-1，当我们调用了其setValue或者postValue，其vesion会+1；对于每一个观察者的封装ObserverWrapper，其初始version也为-1，也就是说，每一个新注册的观察者，其version为-1；当LiveData设置这个ObserverWrapper的时候，如果LiveData的version大于ObserverWrapper的version，LiveData就会强制把当前value推送给Observer。

**如何解决这个问题**

明白了问题产生的原因之后，我们来看看怎么才能解决这个问题。很显然，根据之前的分析，只需要在注册一个新的订阅者的时候把Wrapper的version设置成跟LiveData的version一致即可。

那么怎么实现呢，看看LiveData的observe方法，他会在步骤1创建一个LifecycleBoundObserver，LifecycleBoundObserver是ObserverWrapper的派生类。然后会在步骤2把这个LifecycleBoundObserver放入一个私有Map容器mObservers中。无论ObserverWrapper还是LifecycleBoundObserver都是私有的或者包可见的，所以无法通过继承的方式更改LifecycleBoundObserver的version。

那么能不能从Map容器mObservers中取到LifecycleBoundObserver，然后再更改version呢？答案是肯定的，通过查看SafeIterableMap的源码我们发现有一个protected的get方法。因此，在调用observe的时候，我们可以通过反射拿到LifecycleBoundObserver，再把LifecycleBoundObserver的version设置成和LiveData一致即可。

![640?wx\_fmt=png][640_wx_fmt_png 7]

对于非生命周期感知的observeForever方法来说，实现的思路是一致的，但是具体的实现略有不同。observeForever的时候，生成的wrapper不是LifecycleBoundObserver，而是AlwaysActiveObserver（步骤1），而且我们也没有机会在observeForever调用完成之后再去更改AlwaysActiveObserver的version，因为在observeForever方法体内，步骤3的语句，回调就发生了。

![640?wx\_fmt=png][640_wx_fmt_png 8]

那么对于observeForever，如何解决这个问题呢？既然是在调用内回调的，那么我们可以写一个ObserverWrapper，把真正的回调给包装起来。把ObserverWrapper传给observeForever，那么在回调的时候我们去检查调用栈，如果回调是observeForever方法引起的，那么就不回调真正的订阅者。

**LiveDataBus最终实现**

public final class LiveDataBus {    private final Map<String, BusMutableLiveData<Object>> bus;    private LiveDataBus() {        bus = new HashMap<>();    }    private static class SingletonHolder {        private static final LiveDataBus DEFAULT_BUS = new LiveDataBus();    }    public static LiveDataBus get() {        return SingletonHolder.DEFAULT_BUS;    }    public <T> MutableLiveData<T> with(String key, Class<T> type) {        if (!bus.containsKey(key)) {            bus.put(key, new BusMutableLiveData<>());        }        return (MutableLiveData<T>) bus.get(key);    }    public MutableLiveData<Object> with(String key) {        return with(key, Object.class);    }    private static class ObserverWrapper<T> implements Observer<T> {        private Observer<T> observer;        public ObserverWrapper(Observer<T> observer) {            this.observer = observer;        }        @Override        public void onChanged(@Nullable T t) {            if (observer != null) {                if (isCallOnObserve()) {                    return;                }                observer.onChanged(t);            }        }        private boolean isCallOnObserve() {            StackTraceElement[] stackTrace = Thread.currentThread().getStackTrace();            if (stackTrace != null && stackTrace.length > 0) {                for (StackTraceElement element : stackTrace) {                    if ("android.arch.lifecycle.LiveData".equals(element.getClassName()) &&                            "observeForever".equals(element.getMethodName())) {                        return true;                    }                }            }            return false;        }    }    private static class BusMutableLiveData<T> extends MutableLiveData<T> {        private Map<Observer, Observer> observerMap = new HashMap<>();        @Override        public void observe(@NonNull LifecycleOwner owner, @NonNull Observer<T> observer) {            super.observe(owner, observer);            try {                hook(observer);            } catch (Exception e) {                e.printStackTrace();            }        }        @Override        public void observeForever(@NonNull Observer<T> observer) {            if (!observerMap.containsKey(observer)) {                observerMap.put(observer, new ObserverWrapper(observer));            }            super.observeForever(observerMap.get(observer));        }        @Override        public void removeObserver(@NonNull Observer<T> observer) {            Observer realObserver = null;            if (observerMap.containsKey(observer)) {                realObserver = observerMap.remove(observer);            } else {                realObserver = observer;            }            super.removeObserver(realObserver);        }        private void hook(@NonNull Observer<T> observer) throws Exception {            //get wrapper's version            Class<LiveData> classLiveData = LiveData.class;            Field fieldObservers = classLiveData.getDeclaredField("mObservers");            fieldObservers.setAccessible(true);            Object objectObservers = fieldObservers.get(this);            Class<?> classObservers = objectObservers.getClass();            Method methodGet = classObservers.getDeclaredMethod("get", Object.class);            methodGet.setAccessible(true);            Object objectWrapperEntry = methodGet.invoke(objectObservers, observer);            Object objectWrapper = null;            if (objectWrapperEntry instanceof Map.Entry) {                objectWrapper = ((Map.Entry) objectWrapperEntry).getValue();            }            if (objectWrapper == null) {                throw new NullPointerException("Wrapper can not be bull!");            }            Class<?> classObserverWrapper = objectWrapper.getClass().getSuperclass();            Field fieldLastVersion = classObserverWrapper.getDeclaredField("mLastVersion");            fieldLastVersion.setAccessible(true);            //get livedata's version            Field fieldVersion = classLiveData.getDeclaredField("mVersion");            fieldVersion.setAccessible(true);            Object objectVersion = fieldVersion.get(this);            //set wrapper's version            fieldLastVersion.set(objectWrapper, objectVersion);        }    }}

**注册订阅：**

LiveDataBus.get()        .with("key_test", String.class)        .observe(this, new Observer<String>() {            @Override            public void onChanged(@Nullable String s) {            }        });

**发送消息：**

LiveDataBus.get().with("key_test").setValue(s);

**源码说明**

LiveDataBus的源码可以直接拷贝使用，也可以前往作者的GitHub仓库查看下载：  
[https://github.com/JeremyLiao/LiveDataBus][https_github.com_JeremyLiao_LiveDataBus]

**总结**

本文提供了一个新的消息总线框架——LiveDataBus。订阅者可以订阅某个消息通道的消息，发布者可以把消息发布到消息通道上。利用LiveDataBus，不仅可以实现消息总线功能，而且对于订阅者，他们不需要关心何时取消订阅，极大减少了因为忘记取消订阅造成的内存泄漏风险。

**作者简介**

海亮，美团高级工程师，2017年加入美团，目前主要负责美团轻收银、美团收银零售版等App的相关业务及模块开发工作。

\---------- END ----------

**活动推荐**

![640?wx\_fmt=jpeg][640_wx_fmt_jpeg]

《**美团技术沙龙第40期·北京：前端遇上黑科技，打造全新界面体验与效率**》8月4日周六下午，美团联合百度技术专家为大家揭秘如何让 Hybrid 项目和 Native 体验一致？如何让项目只需要交互稿便可以开始研发？如何让页面首帧速度优化 70% 以上？如何让Node.js 在百度扛起百亿级访问量？更多活动详情请戳>>活动[**报名链接**][Link 1]

**也许你还想看**

[Android自动化页面测速在美团的实践][Android]

[美团外卖Android Crash治理之路][Android Crash]

[Android插件化、热补丁中绕不开的ProGuard的坑][Android_ProGuard]

![640?wx\_fmt=png][640_wx_fmt_png 9]

[640_wx_fmt_png]: https://image.dandelioncloud.cn/pgy_files/images/2024/01/29/757409117b77451cb08e48e293a16697.png
[640_wx_fmt_png 1]: https://image.dandelioncloud.cn/pgy_files/images/2024/01/29/a8ff8b069b89473f871701679829c7e8.png
[640_wx_fmt_png 2]: https://image.dandelioncloud.cn/pgy_files/images/2024/01/29/cefd76dd04f749f5908b5603766c501a.png
[https_github.com_AndroidKnife_RxBus]: https://github.com/AndroidKnife/RxBus
[https_github.com_Blankj_RxBus]: https://github.com/Blankj/RxBus
[640_wx_fmt_png 3]: https://image.dandelioncloud.cn/pgy_files/images/2024/01/29/096e1cc3c814427eb15dcc1139876a25.png
[640_wx_fmt_png 4]: https://image.dandelioncloud.cn/pgy_files/images/2024/01/29/198d449531154497a0ee355c4582b1f7.png
[640_wx_fmt_png 5]: https://image.dandelioncloud.cn/pgy_files/images/2024/01/29/a63e61fd9b774405bcbf603425b86a8e.png
[640_wx_fmt_png 6]: https://image.dandelioncloud.cn/pgy_files/images/2024/01/29/5cba0e6e479843438d42aa4b5928edfe.png
[640_wx_fmt_png 7]: https://image.dandelioncloud.cn/pgy_files/images/2024/01/29/60049fa8b7cd4bf2843e530bc814865f.png
[640_wx_fmt_png 8]: https://image.dandelioncloud.cn/pgy_files/images/2024/01/29/9fe67408f5bf45f2ad0c5b326584bb56.png
[https_github.com_JeremyLiao_LiveDataBus]: https://github.com/JeremyLiao/LiveDataBus
[640_wx_fmt_jpeg]: https://image.dandelioncloud.cn/pgy_files/images/2024/01/29/93706361a86847689334bb864c990ea0.png
[Link 1]: http://www.huodongxing.com/event/3449419913500?qd=weixin
[Android]: http://mp.weixin.qq.com/s?__biz=MjM5NjQ5MTI5OA==&mid=2651748367&idx=2&sn=3e48cdc324eebd56beee6b6b25c74851&chksm=bd12a1428a6528544889254e22c7df7649934d1121ae9f37039ff8b9ff70d278b85c53f24e65&scene=21#wechat_redirect
[Android Crash]: http://mp.weixin.qq.com/s?__biz=MjM5NjQ5MTI5OA==&mid=2651748107&idx=1&sn=55dff1b286e92cfb6aaee776df8ec89e&chksm=bd12ae468a652750a7624c30eca56f6f83347b16cdfb9153b647c6e5229a822b16724a1bbd9d&scene=21#wechat_redirect
[Android_ProGuard]: http://mp.weixin.qq.com/s?__biz=MjM5NjQ5MTI5OA==&mid=2651747907&idx=1&sn=19fd43110781fb702b1dc11d28106398&chksm=bd12af0e8a65261889d1e42f245c355edbae9844575cc62e5d06cd7bdfbbbf90afbd3771e70e&scene=21#wechat_redirect
[640_wx_fmt_png 9]: https://image.dandelioncloud.cn/pgy_files/images/2024/01/29/4748182abc7a44bb8f554f963c45f033.png