如何用LiveDataBus替代RxBus

這篇文章將為大家詳細(xì)講解有關(guān)如何用LiveDataBus替代RxBus，文章內(nèi)容質(zhì)量較高，因此小編分享給大家做個(gè)參考，希望大家閱讀完這篇文章后對(duì)相關(guān)知識(shí)有一定的了解。

對(duì)于Android系統(tǒng)來(lái)說(shuō)，消息傳遞是最基本的組件，每一個(gè)App內(nèi)的不同頁(yè)面，不同組件都在進(jìn)行消息傳遞。消息傳遞既可以用于Android四大組件之間的通信，也可用于異步線程和主線程之間的通信。對(duì)于Android開(kāi)發(fā)者來(lái)說(shuō)，經(jīng)常使用的消息傳遞方式有很多種，從最早使用的Handler、BroadcastReceiver、接口回調(diào)，到近幾年流行的通信總線類框架EventBus、RxBus。Android消息傳遞框架，總在不斷的演進(jìn)之中。

從EventBus說(shuō)起

EventBus是一個(gè)Android事件發(fā)布/訂閱框架，通過(guò)解耦發(fā)布者和訂閱者簡(jiǎn)化Android事件傳遞。EventBus可以代替Android傳統(tǒng)的Intent、Handler、Broadcast或接口回調(diào)，在Fragment、Activity、Service線程之間傳遞數(shù)據(jù)，執(zhí)行方法。

EventBus最大的特點(diǎn)就是簡(jiǎn)潔、解耦。在沒(méi)有EventBus之前我們通常用廣播來(lái)實(shí)現(xiàn)監(jiān)聽(tīng)，或者自定義接口函數(shù)回調(diào)，有的場(chǎng)景我們也可以直接用Intent攜帶簡(jiǎn)單數(shù)據(jù)，或者在線程之間通過(guò)Handler處理消息傳遞。但無(wú)論是廣播還是Handler機(jī)制遠(yuǎn)遠(yuǎn)不能滿足我們高效的開(kāi)發(fā)。EventBus簡(jiǎn)化了應(yīng)用程序內(nèi)各組件間、組件與后臺(tái)線程間的通信。EventBus一經(jīng)推出，便受到廣大開(kāi)發(fā)者的推崇。

現(xiàn)在看來(lái)，EventBus給Android開(kāi)發(fā)者世界帶來(lái)了一種新的框架和思想，就是消息的發(fā)布和訂閱。這種思想在其后很多框架中都得到了應(yīng)用。

圖片摘自EventBus GitHub主頁(yè)

發(fā)布/訂閱模式

訂閱發(fā)布模式定義了一種“一對(duì)多”的依賴關(guān)系，讓多個(gè)訂閱者對(duì)象同時(shí)監(jiān)聽(tīng)某一個(gè)主題對(duì)象。這個(gè)主題對(duì)象在自身狀態(tài)變化時(shí)，會(huì)通知所有訂閱者對(duì)象，使它們能夠自動(dòng)更新自己的狀態(tài)。

RxBus的出現(xiàn)

RxBus不是一個(gè)庫(kù)，而是一個(gè)文件，實(shí)現(xiàn)只有短短30行代碼。RxBus本身不需要過(guò)多分析，它的強(qiáng)大完全來(lái)自于它基于的RxJava技術(shù)。響應(yīng)式編程（Reactive Programming）技術(shù)這幾年特別火，RxJava是它在Java上的實(shí)作。RxJava天生就是發(fā)布/訂閱模式，而且很容易處理線程切換。所以，RxBus憑借區(qū)區(qū)30行代碼，就敢挑戰(zhàn)EventBus“江湖老大”的地位。

RxBus原理

在RxJava中有個(gè)Subject類，它繼承Observable類，同時(shí)實(shí)現(xiàn)了Observer接口，因此Subject可以同時(shí)擔(dān)當(dāng)訂閱者和被訂閱者的角色，我們使用Subject的子類PublishSubject來(lái)創(chuàng)建一個(gè)Subject對(duì)象（PublishSubject只有被訂閱后才會(huì)把接收到的事件立刻發(fā)送給訂閱者），在需要接收事件的地方，訂閱該Subject對(duì)象，之后如果Subject對(duì)象接收到事件，則會(huì)發(fā)射給該訂閱者，此時(shí)Subject對(duì)象充當(dāng)被訂閱者的角色。

完成了訂閱，在需要發(fā)送事件的地方將事件發(fā)送給之前被訂閱的Subject對(duì)象，則此時(shí)Subject對(duì)象作為訂閱者接收事件，然后會(huì)立刻將事件轉(zhuǎn)發(fā)給訂閱該Subject對(duì)象的訂閱者，以便訂閱者處理相應(yīng)事件，到這里就完成了事件的發(fā)送與處理。

最后就是取消訂閱的操作了，RxJava中，訂閱操作會(huì)返回一個(gè)Subscription對(duì)象，以便在合適的時(shí)機(jī)取消訂閱，防止內(nèi)存泄漏，如果一個(gè)類產(chǎn)生多個(gè)Subscription對(duì)象，我們可以用一個(gè)CompositeSubscription存儲(chǔ)起來(lái)，以進(jìn)行批量的取消訂閱。

RxBus有很多實(shí)現(xiàn)，如：

AndroidKnife/RxBus（ https://github.com/AndroidKnife/RxBus ）
Blankj/RxBus（ https://github.com/Blankj/RxBus ）

其實(shí)正如前面所說(shuō)的，RxBus的原理是如此簡(jiǎn)單，我們自己都可以寫(xiě)出一個(gè)RxBus的實(shí)現(xiàn)：

基于RxJava1的RxBus實(shí)現(xiàn)：

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;
    }
    /*
     * 發(fā)送
     */
    public void post(Object o) {
        bus.onNext(o);
    }
    /*
     * 是否有Observable訂閱
     */
    public boolean hasObservable() {
        return bus.hasObservers();
    }
    /*
     * 轉(zhuǎn)換為特定類型的Obserbale
     */
    public <T> Observable<T> toObservable(Class<T> type) {
        return bus.ofType(type);
    }
}

基于RxJava2的RxBus實(shí)現(xiàn)：

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是一個(gè)可以被觀察的數(shù)據(jù)持有類，它可以感知并遵循Activity、Fragment或Service等組件的生命周期。正是由于LiveData對(duì)組件生命周期可感知特點(diǎn)，因此可以做到僅在組件處于生命周期的激活狀態(tài)時(shí)才更新UI數(shù)據(jù)。

LiveData需要一個(gè)觀察者對(duì)象，一般是Observer類的具體實(shí)現(xiàn)。當(dāng)觀察者的生命周期處于STARTED或RESUMED狀態(tài)時(shí)，LiveData會(huì)通知觀察者數(shù)據(jù)變化；在觀察者處于其他狀態(tài)時(shí)，即使LiveData的數(shù)據(jù)變化了，也不會(huì)通知。

LiveData的優(yōu)點(diǎn)

• UI和實(shí)時(shí)數(shù)據(jù)保持一致，因?yàn)長(zhǎng)iveData采用的是觀察者模式，這樣一來(lái)就可以在數(shù)據(jù)發(fā)生改變時(shí)獲得通知，更新UI。

• 避免內(nèi)存泄漏，觀察者被綁定到組件的生命周期上，當(dāng)被綁定的組件銷毀（destroy）時(shí)，觀察者會(huì)立刻自動(dòng)清理自身的數(shù)據(jù)。

• 不會(huì)再產(chǎn)生由于Activity處于stop狀態(tài)而引起的崩潰，例如：當(dāng)Activity處于后臺(tái)狀態(tài)時(shí)，是不會(huì)收到LiveData的任何事件的。

• 不需要再解決生命周期帶來(lái)的問(wèn)題，LiveData可以感知被綁定的組件的生命周期，只有在活躍狀態(tài)才會(huì)通知數(shù)據(jù)變化。

• 實(shí)時(shí)數(shù)據(jù)刷新，當(dāng)組件處于活躍狀態(tài)或者從不活躍狀態(tài)到活躍狀態(tài)時(shí)總是能收到最新的數(shù)據(jù)。

• 解決Configuration Change問(wèn)題，在屏幕發(fā)生旋轉(zhuǎn)或者被回收再次啟動(dòng)，立刻就能收到最新的數(shù)據(jù)。

談一談Android Architecture Components

Android Architecture Components的核心是Lifecycle、LiveData、ViewModel 以及 Room，通過(guò)它可以非常優(yōu)雅的讓數(shù)據(jù)與界面進(jìn)行交互，并做一些持久化的操作，高度解耦，自動(dòng)管理生命周期，而且不用擔(dān)心內(nèi)存泄漏的問(wèn)題。

• Room 
  一個(gè)強(qiáng)大的SQLite對(duì)象映射庫(kù)。

• ViewModel
  一類對(duì)象，它用于為UI組件提供數(shù)據(jù)，在設(shè)備配置發(fā)生變更時(shí)依舊可以存活。

• LiveData 一個(gè)可感知生命周期、可被觀察的數(shù)據(jù)容器，它可以存儲(chǔ)數(shù)據(jù)，還會(huì)在數(shù)據(jù)發(fā)生改變時(shí)進(jìn)行提醒。

• Lifecycle
  包含LifeCycleOwer和LifecycleObserver，分別是生命周期所有者和生命周期感知者。

Android Architecture Components的特點(diǎn)

• 數(shù)據(jù)驅(qū)動(dòng)型編程
  變化的永遠(yuǎn)是數(shù)據(jù)，界面無(wú)需更改。

• 感知生命周期，防止內(nèi)存泄漏

• 高度解耦
  數(shù)據(jù)，界面高度分離。

• 數(shù)據(jù)持久化
  數(shù)據(jù)、ViewModel不與 UI的生命周期掛鉤，不會(huì)因?yàn)榻缑娴闹亟ǘN毀。

重點(diǎn)：為什么使用LiveData構(gòu)建數(shù)據(jù)通信總線LiveDataBus

使用LiveData的理由

• LiveData具有的這種可觀察性和生命周期感知的能力，使其非常適合作為Android通信總線的基礎(chǔ)構(gòu)件。

• 使用者不用顯示調(diào)用反注冊(cè)方法。
  由于LiveData具有生命周期感知能力，所以LiveDataBus只需要調(diào)用注冊(cè)回調(diào)方法，而不需要顯示的調(diào)用反注冊(cè)方法。這樣帶來(lái)的好處不僅可以編寫(xiě)更少的代碼，而且可以完全杜絕其他通信總線類框架（如EventBus、RxBus）忘記調(diào)用反注冊(cè)所帶來(lái)的內(nèi)存泄漏的風(fēng)險(xiǎn)。

為什么要用LiveDataBus替代EventBus和RxBus

• LiveDataBus的實(shí)現(xiàn)極其簡(jiǎn)單，相對(duì)EventBus復(fù)雜的實(shí)現(xiàn)，LiveDataBus只需要一個(gè)類就可以實(shí)現(xiàn)。

• LiveDataBus可以減小APK包的大小，由于LiveDataBus只依賴Android官方Android Architecture Components組件的LiveData，沒(méi)有其他依賴，本身實(shí)現(xiàn)只有一個(gè)類。作為比較，EventBus JAR包大小為57kb，RxBus依賴RxJava和RxAndroid，其中RxJava2包大小2.2MB，RxJava1包大小1.1MB，RxAndroid包大小9kb。使用LiveDataBus可以大大減小APK包的大小。

• LiveDataBus依賴方支持更好，LiveDataBus只依賴Android官方Android Architecture Components組件的LiveData，相比RxBus依賴的RxJava和RxAndroid，依賴方支持更好。

• LiveDataBus具有生命周期感知，LiveDataBus具有生命周期感知，在Android系統(tǒng)中使用調(diào)用者不需要調(diào)用反注冊(cè)，相比EventBus和RxBus使用更為方便，并且沒(méi)有內(nèi)存泄漏風(fēng)險(xiǎn)。

LiveDataBus的設(shè)計(jì)和架構(gòu)

LiveDataBus的組成

• 消息
  消息可以是任何的Object，可以定義不同類型的消息，如Boolean、String。也可以定義自定義類型的消息。

• 消息通道
  LiveData扮演了消息通道的角色，不同的消息通道用不同的名字區(qū)分，名字是String類型的，可以通過(guò)名字獲取到一個(gè)LiveData消息通道。

• 消息總線
  消息總線通過(guò)單例實(shí)現(xiàn)，不同的消息通道存放在一個(gè)HashMap中。

• 訂閱
  訂閱者通過(guò)getChannel獲取消息通道，然后調(diào)用observe訂閱這個(gè)通道的消息。

• 發(fā)布
  發(fā)布者通過(guò)getChannel獲取消息通道，然后調(diào)用setValue或者postValue發(fā)布消息。

LiveDataBus原理圖

LiveDataBus的實(shí)現(xiàn)

第一個(gè)實(shí)現(xiàn)：

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);
    }
}

短短二十行代碼，就實(shí)現(xiàn)了一個(gè)通信總線的全部功能，并且還具有生命周期感知功能，并且使用起來(lái)也及其簡(jiǎn)單：

注冊(cè)訂閱：

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

發(fā)送消息：

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

我們發(fā)送了一個(gè)名為"key_test"，值為true的事件。
這個(gè)時(shí)候訂閱者就會(huì)收到消息，并作相應(yīng)的處理，非常簡(jiǎn)單。

問(wèn)題出現(xiàn)

對(duì)于LiveDataBus的第一版實(shí)現(xiàn)，我們發(fā)現(xiàn)，在使用這個(gè)LiveDataBus的過(guò)程中，訂閱者會(huì)收到訂閱之前發(fā)布的消息。對(duì)于一個(gè)消息總線來(lái)說(shuō)，這是不可接受的。無(wú)論EventBus或者RxBus，訂閱方都不會(huì)收到訂閱之前發(fā)出的消息。對(duì)于一個(gè)消息總線，LiveDataBus必須要解決這個(gè)問(wèn)題。

問(wèn)題分析

怎么解決這個(gè)問(wèn)題呢？先分析下原因：

當(dāng)LifeCircleOwner的狀態(tài)發(fā)生變化的時(shí)候，會(huì)調(diào)用LiveData.ObserverWrapper的activeStateChanged函數(shù)，如果這個(gè)時(shí)候ObserverWrapper的狀態(tài)是active，就會(huì)調(diào)用LiveData的dispatchingValue。

在LiveData的dispatchingValue中，又會(huì)調(diào)用LiveData的considerNotify方法。

在LiveData的considerNotify方法中，紅框中的邏輯是關(guān)鍵，如果ObserverWrapper的mLastVersion小于LiveData的mVersion，就會(huì)去回調(diào)mObserver的onChanged方法。而每個(gè)新的訂閱者，其version都是-1，LiveData一旦設(shè)置過(guò)其version是大于-1的（每次LiveData設(shè)置值都會(huì)使其version加1），這樣就會(huì)導(dǎo)致LiveDataBus每注冊(cè)一個(gè)新的訂閱者，這個(gè)訂閱者立刻會(huì)收到一個(gè)回調(diào)，即使這個(gè)設(shè)置的動(dòng)作發(fā)生在訂閱之前。

問(wèn)題原因總結(jié)

對(duì)于這個(gè)問(wèn)題，總結(jié)一下發(fā)生的核心原因。對(duì)于LiveData，其初始的version是-1，當(dāng)我們調(diào)用了其setValue或者postValue，其vesion會(huì)+1；對(duì)于每一個(gè)觀察者的封裝ObserverWrapper，其初始version也為-1，也就是說(shuō)，每一個(gè)新注冊(cè)的觀察者，其version為-1；當(dāng)LiveData設(shè)置這個(gè)ObserverWrapper的時(shí)候，如果LiveData的version大于ObserverWrapper的version，LiveData就會(huì)強(qiáng)制把當(dāng)前value推送給Observer。

如何解決這個(gè)問(wèn)題

明白了問(wèn)題產(chǎn)生的原因之后，我們來(lái)看看怎么才能解決這個(gè)問(wèn)題。很顯然，根據(jù)之前的分析，只需要在注冊(cè)一個(gè)新的訂閱者的時(shí)候把Wrapper的version設(shè)置成跟LiveData的version一致即可。

那么怎么實(shí)現(xiàn)呢，看看LiveData的observe方法，他會(huì)在步驟1創(chuàng)建一個(gè)LifecycleBoundObserver，LifecycleBoundObserver是ObserverWrapper的派生類。然后會(huì)在步驟2把這個(gè)LifecycleBoundObserver放入一個(gè)私有Map容器mObservers中。無(wú)論ObserverWrapper還是LifecycleBoundObserver都是私有的或者包可見(jiàn)的，所以無(wú)法通過(guò)繼承的方式更改LifecycleBoundObserver的version。

那么能不能從Map容器mObservers中取到LifecycleBoundObserver，然后再更改version呢？答案是肯定的，通過(guò)查看SafeIterableMap的源碼我們發(fā)現(xiàn)有一個(gè)protected的get方法。因此，在調(diào)用observe的時(shí)候，我們可以通過(guò)反射拿到LifecycleBoundObserver，再把LifecycleBoundObserver的version設(shè)置成和LiveData一致即可。

對(duì)于非生命周期感知的observeForever方法來(lái)說(shuō)，實(shí)現(xiàn)的思路是一致的，但是具體的實(shí)現(xiàn)略有不同。observeForever的時(shí)候，生成的wrapper不是LifecycleBoundObserver，而是AlwaysActiveObserver（步驟1），而且我們也沒(méi)有機(jī)會(huì)在observeForever調(diào)用完成之后再去更改AlwaysActiveObserver的version，因?yàn)樵趏bserveForever方法體內(nèi)，步驟3的語(yǔ)句，回調(diào)就發(fā)生了。

那么對(duì)于observeForever，如何解決這個(gè)問(wèn)題呢？既然是在調(diào)用內(nèi)回調(diào)的，那么我們可以寫(xiě)一個(gè)ObserverWrapper，把真正的回調(diào)給包裝起來(lái)。把ObserverWrapper傳給observeForever，那么在回調(diào)的時(shí)候我們?nèi)z查調(diào)用棧，如果回調(diào)是observeForever方法引起的，那么就不回調(diào)真正的訂閱者。

LiveDataBus最終實(shí)現(xiàn)

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);
        }
    }
}

注冊(cè)訂閱：

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

發(fā)送消息：

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

關(guān)于如何用LiveDataBus替代RxBus就分享到這里了，希望以上內(nèi)容可以對(duì)大家有一定的幫助，可以學(xué)到更多知識(shí)。如果覺(jué)得文章不錯(cuò)，可以把它分享出去讓更多的人看到。