当前位置：首页 > article >正文

Android JetPack组件之LiveData的使用详解

article 2025/1/15 12:14:13

一、背景

对于Android系统来说，消息传递是最基本的组件，每一个App内的不同页面，不同组件都在进行消息传递。消息传递既可以用于Android四大组件之间的通信，也可用于异步线程和主线程之间的通信。对于Android开发者来说，经常使用的消息传递方式有很多种，从最早使用的Handler、BroadcastReceiver、接口回调，到近几年流行的通信总线类框架EventBus、RxBus。Android消息传递框架，总在不断的演进之中。

EventBus的优缺点介绍

优点：开销小，代码更优雅、简洁，解耦发送者和接收者，可动态设置事件处理线程和优先级。
缺点：每个事件必须自定义一个事件类，增加了维护成本，需要在适当的时机主动去取消注册，否则容易引起内存泄漏

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

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

二、LiveData的优点

自动取消订阅避免内存泄漏： 当宿主生命周期进入消亡（DESTROYED）状态时，LiveData 会自动移除观察者，避免内存泄漏；
安全地回调数据： 在宿主生命周期状态低于活跃状态（STAETED）时，LiveData 不会回调数据，避免产生空指针异常或不必要的性能损耗；当宿主生命周期不低于活跃状态（STAETED）时，LiveData 会重新尝试回调数据，确保观察者接收到最新的数据。

当观察者的生命周期处于STARTED或RESUMED状态时，LiveData会通知观察者数据变化；在观察者处于其他状态时，即使LiveData的数据变化了，也不会通知。

UI和实时数据保持一致因为LiveData采用的是观察者模式，这样一来就可以在数据发生改变时获得通知，更新UI。
不会再产生由于Activity处于stop状态而引起的崩溃

例如：当Activity处于后台状态时，是不会收到LiveData的任何事件的。

不需要再解决生命周期带来的问题LiveData可以感知被绑定的组件的生命周期，只有在活跃状态才会通知数据变化。
实时数据刷新当组件处于活跃状态或者从不活跃状态到活跃状态时总是能收到最新的数据。
解决Configuration Change问题在屏幕发生旋转或者被回收再次启动，立刻就能收到最新的数据。

三、LiveData的使用

核心方法

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

3.1 引入依赖

implementation "androidx.lifecycle:lifecycle-viewmodel-ktx:2.6.1" // 适用于 Kotlin

implementation "androidx.lifecycle:lifecycle-livedata-ktx:2.6.1" // 适用于 Kotlin

3.2 发送事件

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

通常发送事件直接调用到了MutableLiveData的setValue或者postValue，MutableLiveData是LiveData的具体实现类。其中，MutableLiveData源码比较简单，暴露出下面四个接口

public MutableLiveData(T value)
public MutableLiveData()
public void postValue(T value)
public void setValue(T value)

public class MutableLiveData<T> extends LiveData<T> {

    /**
     * Creates a MutableLiveData initialized with the given {@code value}.
     *
     * @param value initial value
     */
    public MutableLiveData(T value) {
        super(value);
    }

    /**
     * Creates a MutableLiveData with no value assigned to it.
     */
    public MutableLiveData() {
        super();
    }

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

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

通常可以这样封装一个数据类，提供一个MutableLiveData对象，利用这个数据类对象可以提供setValue和postValue两个发送事件的接口。

class MainTitleViewModel : ViewModel() {
    // 使用 MutableLiveData 来更新数据
    private val liveData = MutableLiveData<String>()

    fun getLiveData(): MutableLiveData<String> {
        return liveData
    }

    fun setValue(newValue: String) {
        liveData.value = newValue
    }

    fun postValue(newValue: String) {
        liveData.postValue(newValue)
    }
}

3.2.1 主线程setValue发送消息

Java代码

private final MutableLiveData<String> liveData = new MutableLiveData<>();
liveData.setValue(value)

Kotlin代码

private val liveData = MutableLiveData<String>()
liveData.value = newValue

3.2.2 子线程postValue发送消息

// 示例：使用 postValue 更新数据
new Thread(new Runnable() {
    @Override
    public void run() {
        viewModel.postValue("This is posted value from another thread!");
    }
}).start();

那么postValue和setValue底层实现上有什么不同呢？可以直接查看一下LiveData.java 源码

先通过锁的方式用一个中间变量mPendingData存储value，然后传递了一个Runnable对象mPostValueRunnable给ArchTaskExecutor执行器去执行。

protected void postValue(T value) {
    boolean postTask;
    synchronized (mDataLock) {
        postTask = mPendingData == NOT_SET;
        mPendingData = value;
    }
    if (!postTask) {
        return;
    }
    ArchTaskExecutor.getInstance().postToMainThread(mPostValueRunnable);
}

看mPostValueRunnable具体实现,也很简单,就是把先前存的data通过mPendingData里取出来,然后调用setValue去发送消息

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

mPostValueRunnable里面其实也是调用了setValue,重点看postToMainThread做了什么。往下找,在子类DefaultTaskExecutor实现了该方法,这里其实是拿到了主线程的handler，然后把之前的runable post出去,这样就实现了切换至线程

@Override
public void postToMainThread(Runnable runnable) {
    if (mMainHandler == null) {
        synchronized (mLock) {
            if (mMainHandler == null) {
                mMainHandler = createAsync(Looper.getMainLooper());
            }
        }
    }
    //noinspection ConstantConditions
    mMainHandler.post(runnable);
}

总结 : postValue其实也是通过的setValue发送消息，不同的是，postValue通过handler进行了线程切换,这样在子线程中发送消息也不会报错,而setValue会直接抛出异常

3.3 接收事件

通过public void observe(@NonNull LifecycleOwner owner, @NonNull Observer<? super T> observer) 这个方法能接收到发过来的事件数据，比如这个时候想要在接收到数据后更新textView控件


private val liveData = MutableLiveData<String>()

liveData.observe(this, Observer(){ value ->

    Log.d(TAG, "Received value: $value")

    titleTextView.text = value

})

四、原理

4.1 粘性事件分发流程

先从 LiveData 注册观察者看起:

@MainThread
public void observe(@NonNull LifecycleOwner owner, @NonNull Observer<? super T> observer) {
    assertMainThread("observe");
    if (owner.getLifecycle().getCurrentState() == DESTROYED) {
        // ignore
        return;
    }
    LifecycleBoundObserver wrapper = new LifecycleBoundObserver(owner, observer);
    ObserverWrapper existing = mObservers.putIfAbsent(observer, wrapper);
    if (existing != null && !existing.isAttachedTo(owner)) {
        throw new IllegalArgumentException("Cannot add the same observer"
                + " with different lifecycles");
    }
    if (existing != null) {
        return;
    }
    owner.getLifecycle().addObserver(wrapper);
}

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

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

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

class LifecycleBoundObserver extends ObserverWrapper implements LifecycleEventObserver {
    @NonNull
    final LifecycleOwner mOwner;

    LifecycleBoundObserver(@NonNull LifecycleOwner owner, Observer<? super T> observer) {
        super(observer);
        mOwner = owner;
    }

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

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

    @Override
    boolean isAttachedTo(LifecycleOwner owner) {
        return mOwner == owner;
    }

    @Override
    void detachObserver() {
        mOwner.getLifecycle().removeObserver(this);
    }
}

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

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

@MainThread
// 注册观察者，不会反注册，需自行维护，没有owner无法管理宿主生命周期
public void observeForever(@NonNull Observer<? super T> observer) {
    assertMainThread("observeForever");
    //把`observer`包装成一个`AlwaysActiveObserver`对象
    AlwaysActiveObserver wrapper = new AlwaysActiveObserver(observer);
   // 将Observer存储到mObservers集合中
    ObserverWrapper existing = mObservers.putIfAbsent(observer, wrapper);
    if (existing instanceof LiveData.LifecycleBoundObserver) {
        throw new IllegalArgumentException("Cannot add the same observer"
                + " with different lifecycles");
    }
    if (existing != null) {
        return;
    }
    // 设置为true则不管宿主是否处于可见状态，一直接收数据
    wrapper.activeStateChanged(true);
}

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

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

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

void activeStateChanged(boolean newActive) {
    // 如果状态一致，则退出
    if (newActive == mActive) {
        return;
    }
    // immediately set active state, so we'd never dispatch anything to inactive
    // owner
    // 立即设置状态，就不会分发给非活跃状态
    mActive = newActive;
    changeActiveCounter(mActive ? 1 : -1);
    if (mActive) {
        // 活跃状态，开始分发数据
        dispatchingValue(this);
    }
}


@MainThread
void changeActiveCounter(int change) {
    // mActiveCount表示活跃状态数量
    int previousActiveCount = mActiveCount;
    // 如果是活跃状态的就通过计数器+1，否则就-1
    mActiveCount += change;
    // 如果之前已经是活跃状态分发过就不再分发了
    if (mChangingActiveState) {
        return;
    }
    // 记录当前为活跃状态
    mChangingActiveState = true;
    try {
        // 只要之前的活跃状态数量不等于当前的活跃状态数量，就需要一直更新
        while (previousActiveCount != mActiveCount) {
            boolean needToCallActive = previousActiveCount == 0 && mActiveCount > 0;
            boolean needToCallInactive = previousActiveCount > 0 && mActiveCount == 0;
            previousActiveCount = mActiveCount;
            if (needToCallActive) {
                onActive();
            } else if (needToCallInactive) {
                onInactive();
            }
        }
    } finally {
        mChangingActiveState = false;
    }
}

如果之前的状态和当前的状态一致则直接退出。
然后会去判断当前的状态是否是活跃状态，只有活跃状态的才会去分发事件。
如果是活跃状态的就通过计数器+1，否则就-1。
只要之前的活跃状态数量不等于当前的活跃状态数量，就需要一直更新。判断previousActiveCount是否为0并且mActiveCount是否大于0，如果满足则说明之前没有活跃的，现在有活跃的，此时就会触发 onActive() 方法。previousActiveCount是否大于0并且mActiveCount是否等于0，如果满足则说明之前有活跃的，现在没有活跃的，此时就会触发 onInactive() 方法。

MediatorLiveData.java

onActive()方法执行涉及到的方法

@CallSuper
@Override
protected void onActive() {
    for (Map.Entry<LiveData<?>, Source<?>> source : mSources) {
        source.getValue().plug();
    }
}

void plug() {
    mLiveData.observeForever(this);
}

@MainThread
public void observeForever(@NonNull Observer<? super T> observer) {
    assertMainThread("observeForever");
    AlwaysActiveObserver wrapper = new AlwaysActiveObserver(observer);
    ObserverWrapper existing = mObservers.putIfAbsent(observer, wrapper);
    if (existing instanceof LiveData.LifecycleBoundObserver) {
        throw new IllegalArgumentException("Cannot add the same observer"
                + " with different lifecycles");
    }
    if (existing != null) {
        return;
    }
    wrapper.activeStateChanged(true);
}

onInactive()方法执行涉及到的方法

@CallSuper
@Override
protected void onInactive() {
    for (Map.Entry<LiveData<?>, Source<?>> source : mSources) {
        source.getValue().unplug();
    }
}

void unplug() {
    mLiveData.removeObserver(this);
}

@MainThread
public void removeObserver(@NonNull final Observer<? super T> observer) {
    assertMainThread("removeObserver");
    ObserverWrapper removed = mObservers.remove(observer);
    if (removed == null) {
        return;
    }
    removed.detachObserver();
    removed.activeStateChanged(false);
}

4.2 普通消息发送流程

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

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


void dispatchingValue(@Nullable ObserverWrapper initiator) {
    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;
}


private void considerNotify(ObserverWrapper observer) {
    if (!observer.mActive) {
        return;
    }
    // Check latest state b4 dispatch. Maybe it changed state but we didn't get the event yet.
    //
    // we still first check observer.active to keep it as the entrance for events. So even if
    // the observer moved to an active state, if we've not received that event, we better not
    // notify for a more predictable notification order.
    if (!observer.shouldBeActive()) {
        observer.activeStateChanged(false);
        return;
    }
    if (observer.mLastVersion >= mVersion) {
        return;
    }
    observer.mLastVersion = mVersion;
    observer.mObserver.onChanged((T) mData);
}

五、拓展

对于LiveDataBus的第一版实现，我们发现，在使用这个LiveDataBus的过程中，订阅者会收到订阅之前发布的消息。对于一个消息总线来说，这是不可接受的。无论EventBus或者RxBus，订阅方都不会收到订阅之前发出的消息。对于一个消息总线，LiveDataBus必须要解决这个问题。

5.1 问题分析

怎么解决这个问题呢？先分析下原因：

当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每注册一个新的订阅者，这个订阅者立刻会收到一个回调，即使这个设置的动作发生在订阅之前。

5.2 问题原因总结

对于这个问题，总结一下发生的核心原因。对于LiveData，其初始的version是-1，当我们调用了其setValue或者postValue，其vesion会+1；对于每一个观察者的封装ObserverWrapper，其初始version也为-1，也就是说，每一个新注册的观察者，其version为-1；当LiveData设置这个ObserverWrapper的时候，如果LiveData的version大于ObserverWrapper的version，LiveData就会强制把当前value推送给Observer。

5.3 如何解决这个问题

明白了问题产生的原因之后，我们来看看怎么才能解决这个问题。很显然，根据之前的分析，只需要在注册一个新的订阅者的时候把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方法引起的，那么就不回调真正的订阅者。

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

参考链接：