关于LiveData全面详解（附事件总线）

前言：缤纷色彩闪出的美丽是因它没有分开每种色彩

前言

MVVM 架构模式中，ViewModel 是不会持有宿主的信息，业务逻辑在 ViewModels 层中完成，而不是在 Activities 或 Fragments 中。LiveData 在里面担任数据驱动的作用：

在这里插入图片描述

以往我们使用 Handler，EventBus，RxjavaBus 进行消息通信，LiveData 也是一个种观察者模式，作用跟 RxJava 类似，是观察数据的类，相比 RxJava，一般配合 Jetpack 组件配合使用。它能够在 Activity、Fragment 和 Service 之中正确的处理生命周期。

一、什么是LiveData

1.介绍

LiveData 组件是 Jetpack 新推出的基于观察者的消息订阅/分发组件，具有宿主（Activity/Fragment）生命周期感知能力。这种感知能力可确保 LiveData 仅分发消息给与活跃状态的观察者，即只有处于活跃状态的观察者才能收到消息。

LiveData 的消息分发机制，是以往 Handler，EventBus，RxjavaBus 无法比拟的，它们不会顾及当前页面是否可见，一股脑的有消息就转发。导致即便应用在后台，页面不可见，还在做一些无用的绘制，计算（微信的消息列表是在可见状态下才会更新列表最新信息的）将有限的资源让给可见的页面使用。

活跃状态：Observer所在宿主处于STARTED，RESUMED状态。

本文需要 Lifecycle 的相关知识作为基础，如果有不了解的可以先看看《Android架构灵魂组件Lifecycle的生命周期机制详解》

基于生命周期，其实就是在当前宿主 LifecycleOnwer 注册一个 Observer，那么宿主每次生命周期的变化，都会回调给观察者的 onStateChange() 方法，即便是刚刚注册的观察者宿主也会回调 onStateChange() 方法，会有一个状态同步的过程，LiveData 也是利用这个能力，巧妙实现了当宿主销毁的时候，自动移除注册进来的 Observer，从而避免了手动移除的麻烦。更不会造成内存泄漏，这个也是它的核心思想。

2.LiveData的特点

使用 LiveData 具有以下几点优势：

确保界面符合数据状态：LiveData 遵循观察者模式。当底层数据发生变化时，LiveData 会通知 Observer 对象。
不会发生内存泄漏：观察者会绑定到 Lifecycle 对象，并在其关联的生命周期遭到销毁后进行自我清理。
不会因 Activity 停止而导致崩溃：如果观察者的生命周期处于非活跃状态（如返回堆栈中的 Activity），它便不会接收任何 LiveData 事件。
不再需要手动处理生命周期：界面组件只是观察相关数据，不会停止或恢复观察。LiveData 将自动管理所有这些操作，因为它在观察时可以感知相关的生命周期状态变化。
数据始终保持最新状态：如果生命周期变为非活跃状态，它会在再次变为活跃状态时接收最新的数据。
适当的配置更改：如果由于配置更改（如设备旋转）而重新创建了 Activity 或 Fragment，它会立即接收最新的可用数据。
共享资源：您可以使用单例模式扩展 LiveData 对象以封装系统服务，以便在应用中共享它们。LiveData 对象连接到系统服务一次，然后需要相应资源的任何观察者只需观察 LiveData 对象。

LiveData不足之处：

粘性事件不支持取消（后面注册的观察者也能接收数据，无法反注册，但有办法解决）。

3.LiveData核心方法

方法名	作用
observe(LifecycleOwner owner, Observer observer)	注册和宿主生命周期关联的观察者， owner当前生命周期的宿主，当宿主销毁了observer能自动解除注册
observeForever(Observer observer)	注册观察者，不会反注册，需自行维护，没有owner无法管理宿主生命周期
setValue(T value)	发送数据，没有活跃的观察者时不分发，只能在主线程
postValue(T value)	setValue一样，但是不受线程限制，内部也是通过handelr.post到主线程，最后还是通过setValue来分发的
onActive()	当且仅当有一个活跃的观察者时会触发
onInactive()	不存在活跃的观察者时会触发

二、LiveData的几种用法

1.MutableLiveData

在使用 LiveData 做消息分发的时候，需要使用这个子类，设计的原因是考虑到单一开闭原则，只有拿到 MutableLiveData 才可以发送消息，LiveData 只能接收消息，避免拿到 LiveData 既能发送消息又能接收消息的混乱使用。

public class MutableLiveData<T> extends LiveData<T> {
    @Override
    public void postValue(T value) {
        super.postValue(value);
    }

    @Override
    public void setValue(T value) {
        super.setValue(value);
    }
}

MutableLiveData 仅仅是把上面两个方法从父类的 protect 改为 public 而已，因为在 LiveData 中无法调用 postValue() 和 setValue()。所以我们在使用 LiveData 做消息分发的时候，我们需要使用它的子类 MutableLiveData。

fun mediatorLiveData() {
    val liveData1 = MutableLiveData<String>()
    val liveData2 = MutableLiveData<String>()
    //在创建一个聚合类MediatorLiveData
    val mediatorLiveData = MediatorLiveData<String>()
    //分别把LiveData合并到mediatorLiveData中
    mediatorLiveData.addSource(liveData1, Observer { data ->
        mediatorLiveData.value = data
    })
    mediatorLiveData.addSource(liveData2, Observer { data ->
        mediatorLiveData.value = data
    })

    //数据监听，一旦liveData1或者LiveData2发送了数据，observer便能观察到，以便统一处理更新
    mediatorLiveData.observe(this, Observer { data ->
        LogUtil.e("mediatorLiveData:$data")
    })

    // 模拟发送数据
    liveData1.postValue("liveData1 苏火火苏火火")
    liveData2.postValue("liveData2 苏火火苏火火")
}

打印数据如下：

/com.sum.tea E/LogUtil: mediatorLiveData: liveData1 苏火火苏火火
/com.sum.tea E/LogUtil: mediatorLiveData: liveData2 苏火火苏火火

3.Transformations.map操作符

Transformations.map 可以对 LiveData 的数据在分发给观察者之前进行转换，并且返回一个新的 LiveData 对象。

fun transformationsMap() {
    val mapLiveData = MutableLiveData<Int>()

    //数据转换
    val transformLiveData: LiveData<Int> = Transformations.map(mapLiveData) { input ->
        input * 2
    }

    //使用转换后生成的transformLiveData去观察数据
    transformLiveData.observe(this) { output ->
        LogUtil.e("transformationsMap 数据转换变化:$output")
    }

    //使用原始的LiveData发送数据
    mapLiveData.value = 10
}

打印数据如下：

/com.sum.tea E/LogUtil: transformationsMap 数据转换变化:20

三、LiveData实现原理

1.粘性事件分发流程

先从 LiveData 注册观察者看起:

#LiveData.java
@MainThread
public void observe(LifecycleOwner owner, Observer<? super T> observer) {
    // 如果宿主是DESTROYED状态则直接退出
    if (owner.getLifecycle().getCurrentState() == DESTROYED) {
        return;
    }
    // 把observer包装一个具有生命周期边界的观察者
    LifecycleBoundObserver wrapper = new LifecycleBoundObserver(owner, observer);
    // 存储到mObservers集合
    ObserverWrapper existing = mObservers.putIfAbsent(observer, wrapper);

    if (existing != null) {
        return;
    }
    // 注册到宿主的生命周期里面
    owner.getLifecycle().addObserver(wrapper);
}

把 observer 包装成了 LifecycleBoundObserver，它是一个具有生命周期边界的观察者，它是 LifecycleEventObserver 的子类，
接着把 LifecycleBoundObserver 存储到 mObservers 集合当中。
最后把 LifecycleBoundObserver 注册到宿主的生命周期里面。

所以 wrapper 就能接收到宿主生命周期变化的事件，当第一次注册进去的时候也会触发状态的同步，也能接收到完整的生命周期事件。

因为后面还要做数据的分发，订阅消息就是把这个 Observer 包装成 LifecycleBoundObserver，然后存储到 mObservers 集合当中，有消息的时候就遍历这个集合去分发。

// 具有生命周期边界能力的Observer
class LifecycleBoundObserver extends ObserverWrapper implements LifecycleEventObserver {

    @Override
    boolean shouldBeActive() {
        // 判断观察者是否处于活跃的状态
        return mOwner.getLifecycle().getCurrentState().isAtLeast(STARTED);
    }

    @Override
    public void onStateChanged(LifecycleOwner source, Lifecycle.Event event) {
        // 判断当前宿主的状态是否为destory
        if (mOwner.getLifecycle().getCurrentState() == DESTROYED) {
            // 主动进行反注册，把Observer移除掉
            removeObserver(mObserver);
            return;
        }
        // 状态改变
        activeStateChanged(shouldBeActive());
    }
}

宿主生命周期每一次事件的通知都会回调到 LifecycleEventObserver 的 onStateChanged()，先判断当前宿主的状态是否为 DESTORYED，如果是则主动进行反注册，把 Observer 移除掉。从而主动避免内存泄漏的问题。

如果不是 DESTORYED，那就说明宿主的状态发生了别的变化，触发 activeStateChanged(shouldBeActive()) 这个方法，会先判断观察者是否处于活跃的状态，只有处于活跃状态的观察者才能接收到数据:

// 注册观察者，不会反注册，需自行维护，没有owner无法管理宿主生命周期
public void observeForever(Observer<? super T> observer) {
    //把`observer`包装成一个`AlwaysActiveObserver`对象
    AlwaysActiveObserver wrapper = new AlwaysActiveObserver(observer);
    // 将Observer存储到mObservers集合中
    ObserverWrapper existing = mObservers.putIfAbsent(observer, wrapper);
    // 设置为tru则不管宿主是否处于可见状态，一直接收数据
    wrapper.activeStateChanged(true);
}

observeForever()：它会把 observer 包装成一个 AlwaysActiveObserver 对象， shouldBeActive() 永远为 true，不管你的宿主是否处于可见状态，这就意味着它可以一直接收数据。

除了 observeForever() 这种情况外，观察者是否处于活跃状态其实就等于宿主是否处于活跃的状态。

如果场景是在后台要处理一些事情，可以使用 observeForever() 注册观察者，但是需要在宿主被销毁的时候取消注册，或者使用传统的 callback 形式。

void activeStateChanged(boolean newActive) {
    // 如果状态一致，则退出
    if (newActive == mActive) {
        return;
    }
    // 立即设置状态，就不会分发给非活跃状态
    mActive = newActive;
    // 非活跃状态，mActiveCount表示活跃状态数量
    boolean wasInactive = LiveData.this.mActiveCount == 0;
    LiveData.this.mActiveCount += mActive ? 1 : -1;
    // 注册第一个观察者并且是活跃状态
    if (wasInactive && mActive) {
        // 只有一个观察者时才会触发
        onActive();
    }
    if (LiveData.this.mActiveCount == 0 && !mActive) {
        // 没有观察者时触发
        onInactive();
    }
    if (mActive) {
        // 活跃状态，开始分发数据
        dispatchingValue(this);
    }
}

首先判断 mActiveCount == 0，如果等于0说明里面的观察者没有一个处于活跃的状态。在注册第一个观察者的时候，活跃观察者的数量肯定是等于0的，当注册了第一个观察者之后，它的状态就会发生变化，变成 mActive，此时就会触发 onActive() 方法，如果没有任何一个观察者就会触发 onInactive() 方法，如果 mActive == true，则说明当前观察者处于活跃状态，它是可以接收数据的。

// 分发数据
void dispatchingValue(ObserverWrapper initiator) {
    // ······
    do {
        mDispatchInvalidated = false;
        if (initiator != null) {
            // 1.把数据分发给自己
            considerNotify(initiator);
            initiator = null;
        } else {
            // 2.有新的数据，把mObservers集合当中所有的观察者遍历分发数据
            for (Iterator<Map.Entry<Observer<? super T>, ObserverWrapper>> iterator =
                    mObservers.iteratorWithAdditions(); iterator.hasNext(); ) {
                considerNotify(iterator.next().getValue());
                if (mDispatchInvalidated) {
                    break;
                }
            }
        }
    } while (mDispatchInvalidated);
    mDispatchingValue = false;
}

// 设置数据
protected void setValue(T value) {
    mVersion++;
    mData = value;
    dispatchingValue(null);
}

如果 initiator 为空，本次分发数据就把数据分发给自己 considerNotify(initiator)；
如果 initiator 不为空，一般是从 setVlue() 过来的，这时候就说明有了新的数据，就会把 mObservers 集合当中所有的观察者遍历分发数据。

无论是新注册的观察者还是 setVlue() 触发的消息分发，都会调用 considerNotify() 方法

// 准备通知更新，宿主在恢复活跃状态时也会执行到这里
private void considerNotify(ObserverWrapper observer) {
    // observer非活跃状态直接退出
    if (!observer.mActive) {
        return;
    }
    // 也许它改变了状态，但我们还没有得到事件
    if (!observer.shouldBeActive()) {
        observer.activeStateChanged(false);
        return;
    }
    // observer的版本大于LiveData版本不会发送数据，避免多次重复发送 
    if (observer.mLastVersion >= mVersion) {
        return;
    }
    // 同步观察者的Version数据
    observer.mLastVersion = mVersion;
    observer.mObserver.onChanged((T) mData);
}

这里是数据最终发送的地方，先判断观察者是否处于活跃状态，如果说观察者不活跃，就会 return，当宿主恢复活跃状态的时候就会触发 onStateChanged()，最后也会到这里。

重点来了，LiveData 的 Version 字段和 Observer 的 Vierson 在刚开始创建的时候都是-1，如果 LiveData 已经发送数据了，它的 Version 字段就会加1，如果这个时候新注册了一个 Observer，那么在触发消息分发的时候，这两个字段就不相等，所以 Observer 就能接收到之前发送的消息，在第一次注册的数据的时候（先发数据，后注册的 Observer 也会收到数据）。

这就是粘性事件，目的是为了避免数据多次重复发送，因为每次生命周期的变化都会走到这里。最后调用 observer.mObserver.onChanged() 回调数据。

在这里插入图片描述

这就是一个新的注册的 Observer 是如何接收到之前发送的数据的流程。

2.普通消息分发流程

普通消息分发流程即调用 postValue()，setValue() 才会触发消息的分发。

postValue() 发送数据的流程:

#LiveData.java
// 不限制线程，主线程，子线程都可以调用
protected void postValue(T value) {
    boolean postTask;
    // 加锁
    synchronized (mDataLock) {
        postTask = mPendingData == NOT_SET;
        mPendingData = value;
    }
    if (!postTask) {
        return;
    }
    // 发送到主线程执行
    ArchTaskExecutor.getInstance().postToMainThread(mPostValueRunnable);
}

因为需要 Handler 把这个消息 post 到主线程里面，所以需要把传递进来的 value 保存 mPendingData，当这条消息被执行的时候，就会触发 mPostValueRunnable，里面实际也是调用 setValue()，把上面存储的 mPendingData 传递进去。

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

无论是从哪里发送的数据，接收的地方始终都会发送在主线程，每发送一条消息 mVersion 增加了1，在消息派发的时候就会和 Observer 的 version 进行对比，防止消息重复发送的问题。

就会调用 dispatchingValue(null) 传递了 null，根据上面的分析，如果参数为 null，就会遍历 mObserver 集合中的观察者去逐一判断是否能把数据分发给他们。

@MainThread
protected void setValue(T value) {
    assertMainThread("setValue");
    mVersion++;
    mData = value;
    // 分发数据
    dispatchingValue(null);
}

setValue() 里面调用 dispatchingValue(null) 进行分发数据，就又回到了上面的流程。

在这里插入图片描述

四、总结

1.粘性事件分发流程

通过 observe(owner,observer) 向 LiveData 注册观察者，并且把 observer 包装成一个 LifecycleBoundObserver，它是一个具有生命周期边界的观察者，因为这个观察者只有当宿主处于 STARTED 或者 RESUMED 状态的它才会接收数据，其他时候它是不会接收数据的。
把包装好的 Observer 注册到 Lifecycle 当中，handlerLifecycleEvent(event) 利用 Lifecycle 能力，它能感知宿主生命周期能力的关键地方。注册时和宿主每次生命周期变化都会回调 onStateChanged() 方法，刚进去的时候会触发方法的同步。
会判断这个事件宿主是否被销毁了，从而主动地把 Observer 从 LiveData 中移除掉，流程结束。如果不是 DESTORY，说明宿主当前的状态发生了变化，它会触发 activeStateChanged(boolean newActive) 方法，它会判断当前 Observer 是否处于活跃的状态，如果宿主的状态为 STARTED，RESUMED 则会分发最新数据到每个观察者。
进而调用 dispatchingValue(ObserverWrapper) 分发数据，如果 ObserverWrapper 为空则分发数据给 liveData 中存储的所有观察者，如果不为空，则分发数据给该 Observer。
considerNotify(ObserverWrapper) 中先判断观察者所在的宿主不活跃，则不分发；接着如果 observer 的 mLastVersion 大于或等于 LiveData 的 mVersion 则不分发，防止重复发送数据；最后通过 observer.mObserver.onChanged((T) mData) 分发数据，同步 mVersion 数据。

那么 LiveData 先发送数据，后注册的 Observer 能接收到数据吗？答案是可以的。

2.普通消息发送流程

postValue() 发送一条数据，它可以在任意线程使用的，里面实际使用了 Handler.post 先把这个事件发送到主线程，然后在调用 setValue() 发送数据；
setValue() 代表着 LiveData 发送数据，每发送一次 mVersion++，另外 LifecycleBoundObserver 中也有一个，它代表这个 Observer 接收了几次数据，在分发数据的时候，这两个 version 会进行比对，防止数据重复发送；
setValue() 里面也会触发 dispatchingValue(ObserverWrapper)，ObserverWrapper 为 null，dispatchingValue() 它会遍历 Observer 集合里面所有观察者，然后逐一调用 considerNotify(ObserverWrapper) 去做消息的分发。

五、使用LiveData打造消息总线

基于 LiveData 打造一款不会内存泄漏不用反注册的消息总线，且支持粘性事件。

// 事件总线
private fun liveDataBus() {
    LiveDataBus.with<String?>("eventName").observeSticky(this, { data ->
        mBinding.tvUserInfo.text = data
        LogUtil.e("事件总线 数据变化:$data")
    }, true)
}

问题：LiveData 默认是支持粘性事件的，而且无法取消。

private void considerNotify(ObserverWrapper observer) {
    //观察者没有处于活跃状态，则不分发
    if (!observer.shouldBeActive()) {
        observer.activeStateChanged(false);
        return;
    }
    //观察者接受的消息的次数>=liveData发送消息的次数，不分发
    //如果之前已经发送过数据，新注册的Observer也能接受到最后一条数据
    if (observer.mLastVersion >= mVersion) {
        return;
    }
    
    //根本原因在于ObserverWrapper的version字段在创建时=-1，没有主动和LiveData字段的mVersion字段对齐
    observer.mLastVersion = mVersion;
    observer.mObserver.onChanged((T) mData);
}

满足上面两个条件才会触发 observer.mObserver.onChanged() 事件的分发：

首先判断 observer.shouldBeActive() 是否处于活跃状态，如果 observer 是根据 observer() 注册的，这个观察者是否处于活跃就等于它的宿主是否处于活跃状态，如果是用 observerForever() 注册的，它就一直处于活跃状态。这里不需要控制它，能很好完成与宿主生命周期相关联。
关键在于 observer.mLastVersion >= mVersion，mLastVersion 和 mVersion 默认值都是为-1，在首次注册流程分发的时候，如果 LiveData 之前发送过数据，mVersion 就会为1，而新注册进来的 Observer 的 mLastVersion 还是-1，上就会往下执行消息分发。这样就导致了新注册的 Observer 也能够接收到之前发送消息的最后一条数据。
但是如果我们在新注册 Observer 的时候主动和 LiveData 的 mVersion 对齐，保持一致，那么就不会继续分发消息，这是根本原因。在于 Observer 的 mLastVersion 的值，但是 LifecycleBoundObserver 这个类我们是访问不到的，我们也不能直接控制 mLastVersion 的值，但是我们可以重新包装 Observer。

//包装StickyObserver，有新的消息会回调onChanged方法，从这里判断是否要分发这条消息
//这只是完成StickyObserver的包装，用于控制事件的分发，但是事件的发送还是依靠LiveData来完成的
internal class StickyObserver<T>(
    liveData: StickyLiveData<T>,
    observer: Observer<T>,
    sticky: Boolean
) : Observer<T> {
    private val mLiveData: StickyLiveData<T>
    private val mObserver: Observer<T>

    //是否开启粘性事件,为false则只能接受到注册之后发送的消息，如果需要接受粘性事件则传true
    private val mSticky: Boolean

    //标记该Observer已经接收几次数据了，过滤老数据防止重复接收
    private var mLastVersion = 0

    init {
        //比如先使用StickLiveData发送了一条数据，StickLiveData#version=1
        //那么当创建WrapperObserver注册进去的时候，需要把它的version和StickLiveData的version保持一致
        mLastVersion = liveData.mVersion
        mLiveData = liveData
        mSticky = sticky
        mObserver = observer
    }

    override fun onChanged(t: T) {
        if (mLastVersion >= mLiveData.mVersion) { //如果相等则说明没有更新的数据要发送
            //但是如果当前Observer是关系粘性事件的，则分发给他
            if (mSticky && mLiveData.mStickyData != null) {
                mObserver.onChanged(mLiveData.mStickyData)
            }
            return
        }
        mLastVersion = mLiveData.mVersion
        mObserver.onChanged(t)
    }
}

扩展 LiveData，支持粘性事件的订阅，分发的 StickyLiveData：

//扩展LiveData，支持粘性事件的订阅，分发的StickyLiveData
class StickyLiveData<T>(private val mEventName: String) : LiveData<T>() {
    var mStickyData: T? = null
    // 版本标记
    var mVersion = 0
    // 事件存储集合
    var mHashMap: ConcurrentHashMap<String, StickyLiveData<T>>? = null

     //调用mVersion++
     //注册一个Observer的时候，把它包装一下，目的是为了让Observer的version和LiveData的version对齐
     //但是LiveData的version字段拿不到，所以需要管理version，在对齐的时候
    override fun setValue(value: T?) {
        mVersion++
        super.setValue(value)
    }

    override fun postValue(value: T?) {
        mVersion++
        super.postValue(value)
    }

    //发送粘性事件,只能在主线程发送数据
    fun setStickData(stickyData: T) {
        mStickyData = stickyData
        value = stickyData
    }

    //发送粘性事件，不受线程限制
    fun postStickData(stickyData: T) {
        mStickyData = stickyData
        postValue(stickyData)
    }

    override fun observe(owner: LifecycleOwner, observer: Observer<in T?>) {
        observeSticky(owner, observer, false)
    }

     //暴露方法，是否关心之前发送的数据,再往宿主上面添加一个addObserver监听生命周期事件，如果是DESTORYED则主动移除LiveData
     //sticky  是否为粘性事件，sticky=true,如果之前存在已经发送数据，那么Observer就会收到之前的粘性事件消息
    fun observeSticky(owner: LifecycleOwner, observer: Observer<in T>, sticky: Boolean) {
        owner.lifecycle.addObserver(LifecycleEventObserver { source, event ->
            if (event == Lifecycle.Event.ON_DESTROY) {
                mHashMap?.remove(mEventName)
            }
        })
        super.observe(owner, StickyObserver(this, observer as Observer<T>, sticky))
    }
}