如何用LiveDataBus替代RxBus

這篇文章將為大家詳細講解有關如何用LiveDataBus替代RxBus，文章內容質量較高，因此小編分享給大家做個參考，希望大家閱讀完這篇文章后對相關知識有一定的了解。

對于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開發者世界帶來了一種新的框架和思想，就是消息的發布和訂閱。這種思想在其后很多框架中都得到了應用。

圖片摘自EventBus GitHub主頁

發布/訂閱模式

訂閱發布模式定義了一種“一對多”的依賴關系，讓多個訂閱者對象同時監聽某一個主題對象。這個主題對象在自身狀態變化時，會通知所有訂閱者對象，使它們能夠自動更新自己的狀態。

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 ）
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的優點

• UI和實時數據保持一致，因為LiveData采用的是觀察者模式，這樣一來就可以在數據發生改變時獲得通知，更新UI。

• 避免內存泄漏，觀察者被綁定到組件的生命周期上，當被綁定的組件銷毀（destroy）時，觀察者會立刻自動清理自身的數據。

• 不會再產生由于Activity處于stop狀態而引起的崩潰，例如：當Activity處于后臺狀態時，是不會收到LiveData的任何事件的。

• 不需要再解決生命周期帶來的問題，LiveData可以感知被綁定的組件的生命周期，只有在活躍狀態才會通知數據變化。

• 實時數據刷新，當組件處于活躍狀態或者從不活躍狀態到活躍狀態時總是能收到最新的數據。

• 解決Configuration Change問題，在屏幕發生旋轉或者被回收再次啟動，立刻就能收到最新的數據。

談一談Android Architecture Components

Android Architecture Components的核心是Lifecycle、LiveData、ViewModel 以及 Room，通過它可以非常優雅的讓數據與界面進行交互，并做一些持久化的操作，高度解耦，自動管理生命周期，而且不用擔心內存泄漏的問題。

• Room 
  一個強大的SQLite對象映射庫。

• ViewModel
  一類對象，它用于為UI組件提供數據，在設備配置發生變更時依舊可以存活。

• LiveData 一個可感知生命周期、可被觀察的數據容器，它可以存儲數據，還會在數據發生改變時進行提醒。

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

Android Architecture Components的特點

• 數據驅動型編程
  變化的永遠是數據，界面無需更改。

• 感知生命周期，防止內存泄漏

• 高度解耦
  數據，界面高度分離。

• 數據持久化
  數據、ViewModel不與 UI的生命周期掛鉤，不會因為界面的重建而銷毀。

重點：為什么使用LiveData構建數據通信總線LiveDataBus

使用LiveData的理由

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

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

為什么要用LiveDataBus替代EventBus和RxBus

• LiveDataBus的實現極其簡單，相對EventBus復雜的實現，LiveDataBus只需要一個類就可以實現。

• LiveDataBus可以減小APK包的大小，由于LiveDataBus只依賴Android官方Android Architecture Components組件的LiveData，沒有其他依賴，本身實現只有一個類。作為比較，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系統中使用調用者不需要調用反注冊，相比EventBus和RxBus使用更為方便，并且沒有內存泄漏風險。

LiveDataBus的設計和架構

LiveDataBus的組成

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

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

• 消息總線
  消息總線通過單例實現，不同的消息通道存放在一個HashMap中。

• 訂閱
  訂閱者通過getChannel獲取消息通道，然后調用observe訂閱這個通道的消息。

• 發布
  發布者通過getChannel獲取消息通道，然后調用setValue或者postValue發布消息。

LiveDataBus原理圖

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。

在LiveData的dispatchingValue中，又會調用LiveData的considerNotify方法。

在LiveData的considerNotify方法中，紅框中的邏輯是關鍵，如果ObserverWrapper的mLastVersion小于LiveData的mVersion，就會去回調mObserver的onChanged方法。而每個新的訂閱者，其version都是-1，LiveData一旦設置過其version是大于-1的（每次LiveData設置值都會使其version加1），這樣就會導致LiveDataBus每注冊一個新的訂閱者，這個訂閱者立刻會收到一個回調，即使這個設置的動作發生在訂閱之前。

問題原因總結

對于這個問題，總結一下發生的核心原因。對于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一致即可。

對于非生命周期感知的observeForever方法來說，實現的思路是一致的，但是具體的實現略有不同。observeForever的時候，生成的wrapper不是LifecycleBoundObserver，而是AlwaysActiveObserver（步驟1），而且我們也沒有機會在observeForever調用完成之后再去更改AlwaysActiveObserver的version，因為在observeForever方法體內，步驟3的語句，回調就發生了。

那么對于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替代RxBus就分享到這里了，希望以上內容可以對大家有一定的幫助，可以學到更多知識。如果覺得文章不錯，可以把它分享出去讓更多的人看到。