绪论

本文是学习了大佬的文章后，自己去动手实践后写的一篇学习笔记。大佬的文章写得比较好，我自己去写未必描述得那么清楚😂，所以本文很多地方都直接引用了大佬的文章。

项目源码：https://github.com/LinYaoTian/LiveDataBus

效果图：

Tip：阅读本文最好对 Jetpack 的 LIfeCycle 和 LiveData 有初步的了解。

引用：

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

基于LiveData实现事件总线思路和方案

Android Jetpack架构组件（二）一文带你了解Lifecycle（使用篇）

Android Jetpack架构组件（四）一文带你了解LiveData(使用篇)

Android Jetpack架构组件（五）一文带你了解LiveData(原理篇)

基本概念

对于 Android 系统来说，消息传递是最基本的组件，每一个 App 内的不同页面，不同的组件都在进行消息传递。

消息传递既可以用于 Android 四大组件之前的通信，也可以用于异步线程和主线程之前的通信。

对于 Android 开发者来说，经常使用的消息传递方式有很多种，从最早的 Handler、BroadcastReceiver 、接口回调，到最近几年的流行的通信总线类框架 EventBus、RxBus。

EventBus

说到 Android 的通信总线类框架就不得不提到 EventBus 。

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

EventBus 最大的特点就是：简洁、解耦。

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

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

image

订阅/发布模式

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

image

订阅/发布模式和观察者模式两者非常相似，个人觉得订阅/发布模式是观察者模式的一种增强版。

两者的区别有：

观察者模式下，观察者和被观察者之前直接存在依赖关系。
而订阅/发布模式下，订阅者和发布者之前多了一个调度中心，避免了订阅者和发布者的依赖关系。

image

详情可以看这篇文章：观察者模式 vs 发布-订阅模式

LiveData

本文是通过 LiveData 去实现 EventBus ，这里先介绍下 LiveData。

LiveData 如同它的名字一样，是一个可观察的数据持有者，和常规的 observable 不同，LiveData 是可以具有生命周期感知的，这意味着它能够在 Activity、Fragment、Service 中正确的处理生命周期。

实际上 LiveData 并不能单独起作用，它依赖于 Jectpack 的 LifeCycle 组件，因为 LifeCycle 已经在 Activity 的父类中被封装好了，在我们使用 LiveData 的过程中是基本是无感知的，所以这里就简单提一下 LifeCycle 好了。

LifeCycle 的作用

管理组件的生命周期。
让第三方业务能在自己内部就能拿到依赖的组件的生命周期，便于及时叫停，避免错过执行时机。

想继续了解的同学可以看下这两篇文章：

下面继续说 LiveData：

image

LiveData 的数据源一般是 ViewModel，也可以是其他可以更新的 LiveData 组件。

一般我们使用 LiveData 的 observe()，当数据更新后，LiveData 会通知它的所有活跃的观察者。与 RxJava 不同的，LiveData 只会通知活跃的观察者，例如 Activity 位于 Destroyed 状态时是不活跃的，因此不会收到通知。

当然我们也可以使用 LiveData 的 observerForever() 方法进行订阅，区别是 observerForever() 不会受到 Activity 等组件的生命周期的影响，只要数据更新就会收到通知。

LiveData 的简单使用：

public class MainActivity extends AppCompatActivity {
   private static final String TAG="MainActivity";
    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_main);
        // 创建 LiveData 对象
        MutableLiveData<String> mutableLiveData  = new MutableLiveData<>();
        // 开始订阅
        mutableLiveData.observe(this, new Observer<String>() {
            @Override
            public void onChanged(@Nullable final String s) {
                Log.d(TAG, "onChanged:"+s);
            }
        });
        // 更新数据，最终会回调上面的 onChange() 方法
        mutableLiveData.postValue("Android进阶三部曲");
    }
}

LiveData 的更多详情可以参考这篇文章Android Jetpack架构组件（四）一文带你了解LiveData(使用篇)

通过 LiveData 实现的优点是什么？

EventBus 是业界知名的通信类总线库，但它也存在许多被人诟病的缺点：

需要手动的注册和反注册，稍不小心可能会造成内存泄露。
使用 EventBus 出错时难以跟踪出错的事件源。
每个事件都要定义一个事件类，容易造成类膨胀。

而通过 LiveData 实现的 LiveDataBus 的具有以下优点：

具有生命周期感知能力，不用手动注册和反注册。
具有唯一的可信事件源。
以字符串区分每一个事件，避免类膨胀。
LiveData 为 Android 官方库，更加可靠。

原理

LiveDataBus 的组成

消息： 消息可以是任何的 Object，可以定义不同类型的消息，如 Boolean、String。也可以定义自定义类型的消息。
消息通道： LiveData 扮演了消息通道的角色，不同的消息通道用不同的名字区分，名字是 String 类型的，可以通过名字获取到一个 LiveData 消息通道。
消息总线： 消息总线通过单例实现，不同的消息通道存放在一个 HashMap 中。
订阅： 订阅者通过 getChannel() 获取消息通道，然后调用 observe() 订阅这个通道的消息。
发布： 发布者通过 getChannel() 获取消息通道，然后调用 setValue() 或者 postValue() 发布消息。

LiveDataBus 原理图

image

简单实现

通过 LiveData 我们可以非常简单的实现一个事件发布/订阅框架：

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

没错，上面短短的 27 行代码我们就实现了一个事件发布/订阅框架！

问题一

LiveData 一时使用一时爽，爽完了之后我们发现这个简易的 LiveDataBus 存在一个问题，就是订阅者会收到订阅之前发布的消息，类似于粘性消息。对于一个消息总线来说，粘性消息和非粘性消息都是必须支持的，下面我们来看一下如何解决这个问题。

我们先看一下为什么会出现这个问题：

LiveData 的订阅方法

android.arch.lifecycle.LiveData

@MainThread
   public void observe(@NonNull LifecycleOwner owner, @NonNull Observer<? super T> observer) {
       assertMainThread("observe");
       // 当前绑定的组件(activity or fragment)状态为DESTROYED的时候, 则会忽视当前的订阅请求
       if (owner.getLifecycle().getCurrentState() == DESTROYED) {
           return;
       }
       // 转为带生命周期感知的观察者包装类
       LifecycleBoundObserver wrapper = new LifecycleBoundObserver(owner, observer);
       ObserverWrapper existing = mObservers.putIfAbsent(observer, wrapper);
       // 对应观察者只能与一个owner绑定
       if (existing != null && !existing.isAttachedTo(owner)) {
           throw new IllegalArgumentException("Cannot add the same observer"
                   + " with different lifecycles");
       }
       if (existing != null) {
           return;
       }
       // lifecycle注册
       owner.getLifecycle().addObserver(wrapper);
   }

可以看到，LiveData 内部会将我们的传入参数包装成 wrapper ，然后存在一个 Map 中，最后通过 LifeCycle 组件添加观察者。

LiveData 的更新数据方法

LiveData 更新数据方式有两个，一个是 setValue() 另一个是 postValue()，这两个方法的区别是，postValue() 在内部会抛到主线程去执行更新数据，因此适合在子线程中使用；而 setValue() 则是直接更新数据。

这里就只看下 setValue() 方法就好了

android.arch.lifecycle.LiveData

@MainThread
protected void setValue(T value) {
    assertMainThread("setValue");
    // 发送版本+1
    mVersion++;
    mData = value;
    // 信息分发
    dispatchingValue(null);
}

记住这里的 mVersion，它本问题关键，每次更新数据都会自增，默认值是 -1。然后我们跟进下 dispatchingValue() 方法：

android.arch.lifecycle.LiveData

void dispatchingValue(@Nullable ObserverWrapper initiator) {
        // mDispatchingValue的判断主要是为了解决并发调用dispatchingValue的情况
        // 当对应数据的观察者在执行的过程中, 如有新的数据变更, 则不会再次通知到观察者
        // 所以观察者内的执行不应进行耗时工作
        if (mDispatchingValue) {
            mDispatchInvalidated = true;
            return;
        }
        mDispatchingValue = true;
        do {
            mDispatchInvalidated = false;
            if (initiator != null) {
                // 这里
                considerNotify(initiator);
                initiator = null;
            } else {
                for (Iterator<Map.Entry<Observer<? super T>, ObserverWrapper>> iterator =
                        mObservers.iteratorWithAdditions(); iterator.hasNext(); ) {
                    // 这里
                    considerNotify(iterator.next().getValue());
                    if (mDispatchInvalidated) {
                        break;
                    }
                }
            }
        } while (mDispatchInvalidated);
        mDispatchingValue = false;
    }

可以看到，无论条件判断，最终都会执行 considerNotify() 方法，所以我们继续跟进：

android.arch.lifecycle.LiveData

private void considerNotify(ObserverWrapper observer) {
        if (!observer.mActive) {
            return;
        }
        if (!observer.shouldBeActive()) {
            observer.activeStateChanged(false);
            return;
        }
            // 判断 version
        if (observer.mLastVersion >= mVersion) {
            return;
        }
        observer.mLastVersion = mVersion;
        //noinspection unchecked
        observer.mObserver.onChanged((T) mData);
    }

终于到了最关键的时候了！！如果 ObserverWrapper 的 mLastVersion 小于 LiveData 的 mVersion，那么就会执行的 onChange() 方法去通知观察者数据已更新。

而 ObserverWrapper.mLastVersion 的默认值是 -1， LiveData 只要更新过数据，mVersion 就肯定会大于 -1，所以订阅者会马上收到订阅之前发布的最新消息！！

问题二

我们看一下 LiveData 的 postValue() 方法的注释：

image

可以看到，注释中说，如果在多线程中同一个时刻，多次调用了 postValue() 方法，只有最后次调用的值会得到更新。也就是此方法是有可能会丢失事件！！！

看下源码：

android.arch.lifecycle.LiveData

    private final Runnable mPostValueRunnable = new Runnable() {
        @Override
        public void run() {
            Object newValue;
            synchronized (mDataLock) {
                newValue = mPendingData;
                // 设置 mPendingData 为 not_set
                mPendingData = NOT_SET;
            }
            setValue((T) newValue);
        }
    };

    protected void postValue(T value) {
        boolean postTask;
        synchronized (mDataLock) {
            // 判断 postTask 是否为 not_set
            postTask = mPendingData == NOT_SET;
            mPendingData = value;
        }
        if (!postTask) {
            return;
        }
        ArchTaskExecutor.getInstance().postToMainThread(mPostValueRunnable);
    }

从上面的源码就可以很容易看出，postValue 只是把传进来的数据先存到 mPendingData，然后往主线程抛一个 Runnable，在这个 Runnable 里面再调用 setValue 来把存起来的值真正设置上去，并回调观察者们。而如果在这个 Runnable 执行前多次 postValue，其实只是改变暂存的值 mPendingData，并不会再次抛另一个 Runnable。

最终实现

明白了问题根源所在，我们就好解决问题了。

解决这个问题的方案有多种，其中美团大佬Android消息总线的演进之路：用LiveDataBus替代RxBus、EventBus 使用的反射的方式修改 LiveData 中的 mVersion 值去实现。还有另一个方案基于LiveData实现事件总线思路和方案，此方案是基于自定义观察者包装类，因为粘性消息最终会调用到 Observer#onChange() 方法，因此我们自定义 Observer 包装类，自己维护实际的订阅消息数，来判断是否需要触发真正的 onChange() 方法。本文使用的是第二种方案。

第一步：自定义 Observer 包装类

为了业务的拓展性，这里我们定义先一个 Base 基类：

internal open class BaseBusObserverWrapper<T>(private val mObserver: Observer<in T>, private val mBusLiveData: BusLiveData<T>) : Observer<T> {

    private val mLastVersion = mBusLiveData.version

    private val TAG = "BaseBusObserverWrapper"

    override fun onChanged(t: T?) {
        Logger.d(TAG,"msg receiver = " + t.toString())
        if (mLastVersion >= mBusLiveData.version){
            // LiveData 的版本号没有更新过，说明并没有新数据，只是因为
            // 当前 Observer 的版本号比 LiveData 低导致的调用 onChange()
            return
        }
        try {
            mObserver.onChanged(t)
        }catch (e:Exception){
            Logger.e(TAG,"error on Observer onChanged() = " + e.message)
        }
    }

    open fun isAttachedTo(owner: LifecycleOwner) = false

}

这里我们保存了 LiveData 的 mVersion 值，每次执行 onChange() 时都先判断一些 LiveData 是否更新过数据，如果没有则不执行观察者的 Observer.onChange() 方法。

然后定义两个子类，BusLifecycleObserver 和 BusAlwaysActiveObserver ，其中 BusLifecycleObserver 用于 LiveData#observer() 方法订阅的事件，而 BusAlwaysActiveObserver 用于 LiveData#observerForever() 方法订阅的事件。

internal class BusLifecycleObserver<T>(private val observer: Observer<in T>, private val owner: LifecycleOwner, private val liveData: BusLiveData<T>)
    : BaseBusObserverWrapper<T>(observer,liveData),LifecycleObserver{

    @OnLifecycleEvent(Lifecycle.Event.ON_DESTROY)
    fun onDestroy(){
        liveData.removeObserver(observer)
        owner.lifecycle.removeObserver(this)
    }
}

对于 BusLifecycleObserver，在生命周期组件处于 Destroyed 时，需要把观察者移除。

internal class BusAlwaysActiveObserver<T>(private val mObserver: Observer<in T>, private val mBusLiveData: BusLiveData<T>)
    : BaseBusObserverWrapper<T>(mObserver, mBusLiveData)

对于 BusAlwaysActiveObserver，观察者不受组件生命周期的影响，因此不需要在组件 Destroyed 时移除。

第二步：自定义 LiveData 类

注意：因为 LiveData 的 getVersion() 是包访问级别的！所以 BusLiveData 必须定义到与 LiveData 同一个包内，即 androidx.lifecycle 包下，因此需要你自己创建一个同名的包并将 BusLiveData 放到里面。

如果你的项目工程没有引入 androidx，也可以使用 v7 包下的 android.arch.lifecycle，方法同理，不过要注意的是 android.arch.lifecycle 包下 LiveData 的方法参数的泛型是没有型变的，因此直接复制这里代码会有点问题，需要你自己根据编译器的提示修改下。

class BusLiveData<T>(private val mKey:String) : MutableLiveData<T>() {

    private val TAG = "BusLiveData"

    private val mObserverMap: MutableMap<Observer<in T>, BaseBusObserverWrapper<T>> = mutableMapOf()

    @MainThread
    override fun observe(owner: LifecycleOwner, observer: Observer<in T>) {
        val exist = mObserverMap.getOrPut(observer,{
            BusLifecycleObserver(observer,owner,this).apply {
                mObserverMap[observer] = this
                owner.lifecycle.addObserver(this)
            }
        })
        super.observe(owner, exist)
        Logger.d(TAG,"observe() called with: owner = [$owner], observer = [$observer]")
    }

    @MainThread
    override fun observeForever(observer: Observer<in T>) {
        super.observeForever(observer)
        val exist = mObserverMap.getOrPut(observer ,{
            BusAlwaysActiveObserver(observer,this).apply {
                mObserverMap[observer] = this
            }
        })
        super.observeForever(exist)
        Logger.d(TAG, "observeForever() called with: observer = [$observer]")
    }

    @MainThread
    fun observeSticky(owner: LifecycleOwner, observer: Observer<T>) {
        super.observe(owner, observer)
        Logger.d(TAG, "observeSticky() called with: owner = [$owner], observer = [$observer]")
    }

    @MainThread
    fun observeStickyForever(observer: Observer<T>){
        super.observeForever(observer)
        Logger.d(TAG, "observeStickyForever() called with: observer = [$observer]")
    }

    @MainThread
    override fun removeObserver(observer: Observer<in T>) {
        val exist = mObserverMap.remove(observer) ?: observer
        super.removeObserver(exist)
        Logger.d(TAG, "removeObserver() called with: observer = [$observer]")
    }

    @MainThread
    override fun removeObservers(owner: LifecycleOwner) {
        mObserverMap.iterator().forEach {
            if (it.value.isAttachedTo(owner)) {
                mObserverMap.remove(it.key)
            }
        }
        super.removeObservers(owner)
        Logger.d(TAG, "removeObservers() called with: owner = [$owner]")
    }
    
    override fun postValue(value: T) {
       mMainHandler.post {
           setValue(value)
       }
    }

    @MainThread
    override fun onInactive() {
        super.onInactive()
        if (!hasObservers()) {
            // 当 LiveData 没有活跃的观察者时，可以移除相关的实例
            LiveDataBusCore.getInstance().mBusMap.remove(mKey)
        }
        Logger.d(TAG, "onInactive() called")
    }

    @MainThread
    public override fun getVersion(): Int {
        return super.getVersion()
    }


}

代码比较短和简单，要点就是，对于非粘性消息，即 observer() 和 observerForever() ，我们需要使用自定义的包装类包装处理。对于粘性消息，则直接使用 LiveData 默认实现即可。

同时重写 postValue() 方法，内部通过 MainHandler.post() 的方式去执行 setValue() 去避免 postValue() 会丢失数据的问题。

第三步：定义管理类

internal class LiveDataBusCore {

    companion object{

        @JvmStatic
        private val defaultBus = LiveDataBusCore()

        @JvmStatic
        fun getInstance() = defaultBus
    }

    internal val mBusMap : MutableMap<String, BusLiveData<*>> by lazy {
        mutableMapOf<String, BusLiveData<*>>()
    }

    fun <T> getChannel(key: String) : BusLiveData<T> {
        return mBusMap.getOrPut(key){
            BusLiveData<T>(key)
        } as BusLiveData<T>
    }
}

第四步：定义入口类

class LiveDataBus {

    companion object{

        @JvmStatic
        @Synchronized
        fun <T> getSyn(key: String) : BusLiveData<T>{
            return get(key)
        }
      
        @JvmStatic
        fun <T> get(key: String) : BusLiveData<T>{
            return LiveDataBusCore.getInstance().getChannel(key)
        }
      
        private fun <T> get(key: String, type: Class<T>) : BusLiveData<T> {
            return LiveDataBusCore.getInstance().getChannel(key)
        }

        @JvmStatic
        fun <E> of(clz: Class<E>): E {
            require(clz.isInterface) { "API declarations must be interfaces." }
            require(clz.interfaces.isEmpty()) { "API interfaces must not extend other interfaces." }
            return Proxy.newProxyInstance(clz.classLoader, arrayOf(clz), InvocationHandler { _, method, _->
                return@InvocationHandler get(
                        // 事件名以集合类名_事件方法名定义
                        // 以此保证事件的唯一性
                        "${clz.canonicalName}_${method.name}",
                        (method.genericReturnType as ParameterizedType).actualTypeArguments[0].javaClass)
            }) as E
        }
    }


}

使用：

class TestLiveDataBusActivity : AppCompatActivity() {
  
    companion object{
        private const val TAG = "TestLiveDataBusActivity"
    }

    override fun onCreate(savedInstanceState: Bundle?) {
        super.onCreate(savedInstanceState)
        setContentView(R.layout.activity_test_live_data_bus)

        LiveDataBus.get<String>("testObserver").observe(this, Observer<String> {
            Log.d(TAG, "testObserver = $it")
            test_observer_id.text = it
        })
        
        LiveDataBus.get<String>("testObserver").postValue("new value")
    }
}

LiveDataBus 还提供了一个 of() 方法，用于提供用于事件约束，什么是时间约束呢？就是说我们的现在观察者和发布者获取消息渠道的 key 是一个字符串，在使用的过程中可能会出现，发布者的 key 是 itO 而观察者的 key 误输入成 it0 的情况，所以这里我们可以模仿 Retrofit 请求动态代理的做法，在使用的过程中，我们需要先定义一个接口：

interface TestLiveEvents {
    fun event1(): MutableLiveData<String>
}

使用：

fun main() {
    LiveDataBus
            .of(TestLiveEvents::class.java)
            .event1()
            .postValue("new value")
}

总结

借助于 Android 官方提供的 LiveData ，我们可以非常方便地实现自己的 LiveDataBus，全部文件也就只有以上几个类，同时我们还避免了 EventBus 的许多缺点！

LiveDataBus 的源码：https://github.com/LinYaoTian/LiveDataBus

如果文章有任何问题，欢迎在评论区指正。

Android 使用 LiveData 实现 EventBus

Android 使用 LiveData 实现 EventBus

绪论

基本概念

EventBus

订阅/发布模式

LiveData

通过 LiveData 实现的优点是什么？

原理

LiveDataBus 的组成

LiveDataBus 原理图

简单实现

问题一

LiveData 的订阅方法

LiveData 的更新数据方法

问题二

最终实现

第一步：自定义 Observer 包装类

第二步：自定义 LiveData 类

第三步：定义管理类

第四步：定义入口类

总结