LiveDataBus：Android消息匯流排的演進之路

來自專欄美團技術博客37 人贊了文章

背景

對於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）

其實正如前面所說的，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 wrappers 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 livedatas version Field fieldVersion = classLiveData.getDeclaredField("mVersion"); fieldVersion.setAccessible(true); Object objectVersion = fieldVersion.get(this); //set wrappers 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倉庫查看下載：

總結

本文提供了一個新的消息匯流排框架——LiveDataBus。訂閱者可以訂閱某個消息通道的消息，發布者可以把消息發布到消息通道上。利用LiveDataBus，不僅可以實現消息匯流排功能，而且對於訂閱者，他們不需要關心何時取消訂閱，極大減少了因為忘記取消訂閱造成的內存泄漏風險。

作者簡介

海亮，美團高級工程師，2017年加入美團，目前主要負責美團輕收銀、美團收銀零售版等App的相關業務及模塊開發工作。

活動推薦

《美團技術沙龍第40期·北京：前端遇上黑科技，打造全新界面體驗與效率》8月4日周六下午，美團聯合百度技術專家為大家揭秘如何讓 Hybrid 項目和 Native 體驗一致？如何讓項目只需要交互稿便可以開始研發？如何讓頁面首幀速度優化 70% 以上？如何讓Node.js 在百度扛起百億級訪問量？更多活動詳情請戳>>活動報名鏈接

也許你還想看

Android自動化頁面測速在美團的實踐

美團外賣Android Crash治理之路

Android插件化、熱補丁中繞不開的ProGuard的坑

http://weixin.qq.com/r/9HVSSg3EOFBHrUkp9yDm (二維碼自動識別)