程序员Android开发Android技术知识

Android消息总线的演进之路:用LiveDataBus替代R

2019-02-27  本文已影响10人  Android高级开发

正文:

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

从 EventBus 说起

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

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

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

Picture

发布/订阅模式

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

Picture

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
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 谈起

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

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

LiveData 的优点

谈一谈 Android Architecture Components

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

一个强大的SQLite对象映射库。

一类对象,它用于为UI组件提供数据,在设备配置发生变更时依旧可以存活。

一个可感知生命周期、可被观察的数据容器,它可以存储数据,还会在数据发生改变时进行提醒。

包含LifeCycleOwerLifecycleObserver,分别是生命周期所有者和生命周期感知者。

Android Architecture Components 的特点

变化的永远是数据,界面无需更改。

数据,界面高度分离。

数据、ViewModel不与 UI的生命周期挂钩,不会因为界面的重建而销毁。

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

使用 LiveData 的理由

由于LiveData具有生命周期感知能力,所以LiveDataBus只需要调用注册回调方法,而不需要显示的调用反注册方法。这样带来的好处不仅可以编写更少的代码,而且可以完全杜绝其他通信总线类框架(如EventBusRxBus)忘记调用反注册所带来的内存泄漏的风险。

为什么要用LiveDataBus替代EventBus和RxBus

LiveDataBus 的设计和架构

LiveDataBus 的组成

消息可以是任何的Object,可以定义不同类型的消息,如BooleanString。也可以定义自定义类型的消息。

LiveData扮演了消息通道的角色,不同的消息通道用不同的名字区分,名字是String类型的,可以通过名字获取到一个LiveData消息通道。

消息总线通过单例实现,不同的消息通道存放在一个HashMap中。

订阅者通过getChannel获取消息通道,然后调用observe订阅这个通道的消息。
发布
发布者通过getChannel获取消息通道,然后调用setValue或者postValue发布消息。
LiveDataBus原理图

Picture

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.ObserverWrapperactiveStateChanged函数,如果这个时候ObserverWrapper的状态是active,就会调用LiveDatadispatchingValue

Picture

LiveDatadispatchingValue中,又会调用LiveDataconsiderNotify方法。

Picture

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

Picture

问题原因总结

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

如何解决这个问题

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

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

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

Picture

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

Picture

那么对于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 &amp;&amp; stackTrace.length > 0) {
                for (StackTraceElement element : stackTrace) {
                    if ("android.arch.lifecycle.LiveData".equals(element.getClassName()) &amp;&amp;
                            "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

总结

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

image image

+qq群:853967238。获取以上高清技术思维图,以及相关技术的免费视频学习资料

上一篇下一篇

猜你喜欢

热点阅读