文章目录
前言
LiveData 是 Jetpack 的组件之一，是一个可感知目标生命周期并且可被观察的数据容器，当容器内数据变化时会根据观察者是否处于活跃状态回调最新数据给观察者，当观察者从非活跃状态变为活跃状态时同样会回调最新数据给观察者，并且当所属组件生命周期变为 DESTROYED 状态时会自动移除观察者避免内存泄漏，下面来分析一下 LiveData 的源码
提示：基于源码 2.4.0
使用
// MutableLiveData 是 LiveData 的子类
// 只是将 setValue() 和 postValue() 方法的访问权限提升为了 public 从而让外部可以直接调用这两个方法
val liveData = MutableLiveData<String>()
liveData.observe(this) {
}
liveData.observeForever {
}
// 主线程调用
liveData.value = "setValue"
// 子线程调用
liveData.postValue("postValue")
LiveData 的使用非常简单，创建一个 MutableLiveData 对象调用 observe 或者 observeForever 添加数据观察者，然后通过 setValue 或者 postValue 改变数据，当数据发生变化时回调给观察者 observe 与 observeForever 的区别是通过 observe 传入的观察者只会在所属组件生命周期处于活跃状态时才会收到数据改变的通知 observeForever 不对所属组件生命周期进行判断一直可以收到数据改变的通知，所以这个方法不需要传入 LifecycleOwner 下面依次看一下这几个方法
除了调用 setValue/postValue 之外 LiveData 观察者所属组件生命周期变化时也可能会触发通知
observe
public void observe(@NonNull LifecycleOwner owner, @NonNull Observer<? super T> observer) {
// 限制在主线程调用
assertMainThread("observe");
// 如果传入的 owner 表示的组件所处的生命周期处于 DESTROYED 状态则忽略
if (owner.getLifecycle().getCurrentState() == DESTROYED) {
// ignore
return;
}
// 将传入的 owner 和 observer 包装成 LifecycleBoundObserver
LifecycleBoundObserver wrapper = new LifecycleBoundObserver(owner, observer);
// 将 observer 作为 key，wrapper 作为 value 进行存储
// 当 mObservers 不包含该 key 时，调用 putIfAbsent 会返回 null 并且存储 key-value
// 当 mObservers 已包含该 key 时，调用 putIfAbsent 不会存储 key-value，并会返回之前保存的 value
ObserverWrapper existing = mObservers.putIfAbsent(observer, wrapper);
// 如果已经包含 observer 并且传入的 owner 与上一次传入的不是同一个则抛出异常
// 这说明一个 observer 只能与一个 owner 绑定，如果与多个 owner 绑定的话数据变化时不知道判断哪个 owner 的生命周期状态了
if (existing != null && !existing.isAttachedTo(owner)) {
throw new IllegalArgumentException("Cannot add the same observer"
+ " with different lifecycles");
}
// 如果已经包含 observer 并且传入的 owner 是同一个则忽略
if (existing != null) {
return;
}
// 添加生命周期的观察者，感知组件的生命周期
owner.getLifecycle().addObserver(wrapper);
}
到这里提到了两种观察者，一种是组件生命周期的观察者，一种是 LiveData（数据或者说数据容器）的观察者
将 LifecycleOwner (表示可以被感知生命周期的组件) 和 Observer 对象封装成 LifecycleBoundObserver 对象，并且将对象加入到 owner 表示的组件生命周期观察者中，它 是 ObserverWrapper 的子类并且实现了 LifecycleEventObserver 接口
private abstract class ObserverWrapper {
// 传入的 LiveData 观察者
final Observer<? super T> mObserver;
// 表示所属组件（实现了 LifecycleOwner 接口的组件）是否处于活跃状态
boolean mActive;
// 用于与 LiveData 中的版本比较判断是否需要通知数据变化
int mLastVersion = START_VERSION;
ObserverWrapper(Observer<? super T> observer) {
mObserver = observer;
}
// 返回所属组件是否处于活跃状态
abstract boolean shouldBeActive();
// 是否与传入的 LifecycleOwner 对象绑定
boolean isAttachedTo(LifecycleOwner owner) {
return false;
}
// 取消感知组件的生命周期
void detachObserver() {
}
// 根据组件状态决定是否需要通知数据变化
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);
}
}
}
ObserverWrapper 是一个抽象类，唯一的抽象方法表示所属组件是否处于活跃状态 LifecycleBoundObserver 继承自 ObserverWrapper 实现了这个方法
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() {
// 所属组件处于 STARTED 或 RESUMED 状态才认为处于活跃状态
return mOwner.getLifecycle().getCurrentState().isAtLeast(STARTED);
}
@Override
// LifecycleEventObserver 接口方法
// 当所属组件生命周期状态变化时回调
public void onStateChanged(@NonNull LifecycleOwner source,
@NonNull Lifecycle.Event event) {
// 所属组件的生命周期状态
Lifecycle.State currentState = mOwner.getLifecycle().getCurrentState();
// 如果组件处于 DESTROYED 状态则移除观察者返回
if (currentState == DESTROYED) {
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);
}
}
LifecycleBoundObserver 重写了 shouldBeActive 方法判断组件是否处于活跃状态，并且实现了 LifecycleEventObserver 接口重写 onStateChanged 方法处理所属组件生命周期变化（所以才可以在 LiveData 观察者绑定的组件生命周期变为活跃状态时收到最新的数据但这些带来了一些问题，后面会分析），如果组件处于 DESTROYED 状态则移除观察者，如果处于其他状态则调用 activeStateChanged 传入 shouldBeActive 方法的返回值
@MainThread
public void removeObserver(@NonNull final Observer<? super T> observer) {
// 限制在主线程调用
assertMainThread("removeObserver");
// 移除 LiveData 观察者
ObserverWrapper removed = mObservers.remove(observer);
if (removed == null) {
return;
}
// 取消观察所属组件的生命周期
removed.detachObserver();
// 同样调用 activeStateChanged 方法传入 false
removed.activeStateChanged(false);
}
不管所属组件生命周期处于什么状态都会调用 activeStateChanged 方法，只不过传入的值可能不一样 activeStateChanged 方法则根据传入的值决定是否需要通知观察者数据变化
void activeStateChanged(boolean newActive) {
// 如果所属组件活跃状态没有变化则不处理
if (newActive == mActive) {
return;
}
// 记录组件活跃状态
mActive = newActive;
// 根据所属组件活跃状态
// 当活跃观察者数量变化时调用相应方法
changeActiveCounter(mActive ? 1 : -1);
if (mActive) {
// 如果所属组件处于活跃状态回调数据改变
dispatchingValue(this);
}
}
@MainThread
void changeActiveCounter(int change) {
// 记录之前处于活跃状态的观察者的数量
int previousActiveCount = mActiveCount;
// 更新处于活跃状态的观察者的数量
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;
}
}
void dispatchingValue(@Nullable ObserverWrapper initiator) {
// 如果正在通知数据变化
if (mDispatchingValue) {
// 如果在通知数据变化的过程中又触发了此方法说明在回调方法中再次改变了数据需要再次通知数据变化（把最新的数据变化通知出去）
mDispatchInvalidated = true;
return;
}
// 标记正在通知数据变化
mDispatchingValue = true;
do {
mDispatchInvalidated = false;
// 这里有两种情况
// initiator 为 null 时是通知所有观察者，对应的情况是 LiveData 有新值到来，外部调用了 setValue 或者 postValue
// 不为 null 时只通知该 ObserverWrapper，对应的情况是外部新添加了一个 Observer 或者所属组件生命周期变化
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;
}
不管是什么场景触发 dispatchingValue 最后都会调用 considerNotify 它才是最终通知观察者数据变化的方法
private void considerNotify(ObserverWrapper observer) {
// 如果所属组件处于不活跃状态直接返回
if (!observer.mActive) {
return;
}
// 此处判断主要是为了照顾 LifecycleBoundObserver
// 由于 Lifecycle 有可能状态值 State 已经切换到了非活跃状态，但 LifecycleBoundObserver 还未收到事件通知
// 所以为了避免意外情况，此处主动检查 observer 的活跃状态并判断是否需要更新其活跃状态
if (!observer.shouldBeActive()) {
observer.activeStateChanged(false);
return;
}
// 版本号判断避免重复通知比如通过 setValue/postValue 之后组件生命周期变化（待会会具体说版本的问题）
if (observer.mLastVersion >= mVersion) {
return;
}
// 观察者记录版本
observer.mLastVersion = mVersion;
// 真正的通知数据变化，调用观察者的 onChanged 方法
observer.mObserver.onChanged((T) mData);
}
到这里就能收到数据变化的通知了，再来看一下 observeForever
上面因为说到生命周期变化的回调所以直接看下来就到了通知观察者的地方，并且 LiveData 之所以粘性也有这部分的原因（下面会分析下粘性），就不再分析这一块了
observeForever
@MainThread
public void observeForever(@NonNull Observer<? super T> observer) {
// 限制在主线程调用
assertMainThread("observeForever");
// AlwaysActiveObserver 是 ObserverWrapper 的另一个子类
// 它的 shouldBeActive 直接返回 true 不判断组件是否处于活跃状态直接通知
AlwaysActiveObserver wrapper = new AlwaysActiveObserver(observer);
ObserverWrapper existing = mObservers.putIfAbsent(observer, wrapper);
// 如果已经通过 observe 添加了观察者则抛出异常
if (existing instanceof LiveData.LifecycleBoundObserver) {
throw new IllegalArgumentException("Cannot add the same observer"
+ " with different lifecycles");
}
// 如果已经通过 observeForever 添加了观察者则忽略
if (existing != null) {
return;
}
// 主动触发 activeStateChanged 如果已经设置过 LiveData 则会收到最新的值
// 参数 true 表示是活跃状态的观察者
wrapper.activeStateChanged(true);
}
private class AlwaysActiveObserver extends ObserverWrapper {
AlwaysActiveObserver(Observer<? super T> observer) {
super(observer);
}
@Override
boolean shouldBeActive() {
return true;
}
}
observeForever 只有一个 LiveData 观察者参数没有传入 LifecycleOwner 所以组件处于任何状态都可能收到数据变化的通知，并且也不会在组件处于 DESTROYED 状态时自动移除观察者需要使用者主动移除。另外同一个 LiveData 观察者值可以调用一次 observe 或者 observeForever 方法，两个都调用会抛出异常，重复调用某一个会忽略之后的调用（对于 observe 如果两次传入的 owner 不一样也会抛出异常）
setValue
setValue 方法用于设置 LiveData 数据变化，只可以在主线程调用，调用后处于活跃状态的观察者就会收到数据改变的通知
@MainThread
protected void setValue(T value) {
assertMainThread("setValue");
// 每次调用版本号加 1
mVersion++;
// 记录最新的值
mData = value;
// 通知所有的观察者数据改变，后续流程就跟生命周期变化后的流程一样了
dispatchingValue(null);
}
postValue
postValue 同样用于设置 LiveData 数据改变可以在任意线程调用，处于活跃状态的观察者就会收到数据改变的通知
final Object mDataLock = new Object();
// mPendingData 的默认值
// 当 mPendingData 等于 NOT_SET 时说明当前 LiveData 没有值需要通过 postValue 回调
static final Object NOT_SET = new Object();
// 记录通过 postValue 设置的最新值
// 因为是多线程添加了 volatile 关键字
volatile Object mPendingData = NOT_SET;
protected void postValue(T value) {
boolean postTask;
synchronized (mDataLock) {
postTask = mPendingData == NOT_SET;
mPendingData = value;
}
// 从发送一个 runnable 到主线程的消息队列到真正去执行这个 ruannble 是有一个时间间隔的
// 如果 postTask 为 false 即 mPendingData != NOT_SET 说明已经通过 postValue 设置了一个值
// 并且已经有一个 mPostValueRunnable 正在主线程的消息队列里等待执行
// 所以不需要再 post 一次 mPostValueRunnable 了
// 只需要修改 mPendingData 的值等 mPostValueRunnable 执行到的时候用的就是最新的值了
if (!postTask) {
return;
}
// 将 mPostValueRunnable 发送到主线程执行
ArchTaskExecutor.getInstance().postToMainThread(mPostValueRunnable);
}
private final Runnable mPostValueRunnable = new Runnable() {
@SuppressWarnings("unchecked")
@Override
public void run() {
Object newValue;
synchronized (mDataLock) {
// 通过 postValue 设置的最新的值
newValue = mPendingData;
// 置为初始值
mPendingData = NOT_SET;
}
// 调用 setValue 传入最新的值
setValue((T) newValue);
}
};
最后走的还是 setValue 的逻辑
粘性事件
LiveData 是粘性的即先改变 LiveData 数据后通过 observe/observeForever 添加观察者，后添加的观察者依然可以接收到数据改变的通知，通过 observeForever 添加的观察者没有关联生命周期但是在添加时主动调用了 activeStateChanged(true) 如果数据改变或则会通知到观察者，而通过 observe 添加的观察者被封装成了 LifecycleBoundObserver 对象（实现了 LifecycleEventObserver 接口实现了 onStateChanged 方法，前面已分析过了）并且被添加到了组件生命周期的观察者里，因为 Lifecycle 状态变化发送的事件本身就是粘性的所以基于 Lifecycle 的通过 observe 方法添加的 LiveData 观察者也是粘性的
还有一个原因当界面被意外销毁后，我们需要根据已有的数据来进行界面重建，所以 LiveData 被设计为黏性的
粘性事件的说法应该是来自 EventBus 的 StickyEvent Jetpack LiveData 的设计理念及改进
// androidx.lifecycle.LifecycleRegistry#addObserver
public void addObserver(@NonNull LifecycleObserver observer) {
// 省略代码 ...
// 依次发送组件生命周期事件（会转换为状态）给新的观察者
while ((statefulObserver.mState.compareTo(targetState) < 0
&& mObserverMap.contains(observer))) {
pushParentState(statefulObserver.mState);
final Event event = Event.upFrom(statefulObserver.mState);
if (event == null) {
throw new IllegalStateException("no event up from " + statefulObserver.mState);
}
// 发送生命周期事件
statefulObserver.dispatchEvent(lifecycleOwner, event);
popParentState();
// mState / subling may have been changed recalculate
targetState = calculateTargetState(observer);
}
// 省略代码 ...
}
// androidx.lifecycle.LifecycleRegistry.ObserverWithState#dispatchEvent
void dispatchEvent(LifecycleOwner owner, Event event) {
State newState = event.getTargetState();
mState = min(mState, newState);
// 对于 LiveData 通过 observe 添加的观察者这里就是 androidx.lifecycle.LiveData.LifecycleBoundObserver#onStateChanged
mLifecycleObserver.onStateChanged(owner, event);
mState = newState;
}
// androidx.lifecycle.LiveData.LifecycleBoundObserver#onStateChanged 最终会调用到
// androidx.lifecycle.LiveData#considerNotify
private void considerNotify(ObserverWrapper observer) {
// 省略代码 ...
// mLastVersion 与 mVersion 初始值都是 -1
// 如果通过 setValue 改变过 LiveData 的数据 mVersion 会 +1
// observer 是新添加的 LiveData 观察者所以 mLastVersion = -1
// 所以 observer.mLastVersion >= mVersion 条件不成立会继续执行
// 最终发送了之前设置的数据给 LiveData 观察者
if (observer.mLastVersion >= mVersion) {
return;
}
observer.mLastVersion = mVersion;
observer.mObserver.onChanged((T) mData);
}
可以看到添加生命周期观察者时回调了之前的事件给观察者，并且由于 LiveData 观察者的版本号默认是 -1 如果 LiveData 在这之前已经改变了数据，版本的判断也拦不住，就会把最新的数据发送出来
上面说了粘性事件的定义是先改变数据后添加观察者依然可以收到通知，但还有一个场景也有类似的问题只是还叫做粘性事件有些不准确，比如在 AB 两个页面使用了同一个 ViewModel 中的 LiveData 先打开页面 A 并且观察了这个 LiveData 弹出一个 Toast 在 A 页面触发某个操作进入了 B 页面也观察了 LiveData 显示一个 Snackbar 提示，在 B 页面通过改变 LiveData 的值，这个时候由于 A 页面处于非活跃状态所以只会通知到 B 页面显示一个 Snackbar 提示，但是当从 B 页面再次回到 A 页面时 A 变成活跃状态并且版本号的判断也没有拦住会在 A 页面弹出一个 Toast 显然这也不是想要的（所以统称为 ‘数据倒灌’ 可能更为合适一些）
‘数据倒灌’ 解决方案
- Event 事件包装器 对于多观察者的情况，只允许第一个观察者消费，这不符合现实需求；而且手写 Event 事件包装器，在 Java 中存在 null 安全的一致性问题。
- 反射干预 Version 的方式 存在延迟，无法用于对实时性有要求的场景；并且数据会随着 SharedViewModel 长久滞留在内存中得不到释放。
- SingleLiveEvent 是对 Event 事件包装器 一致性问题的改进，但未解决多观察者消费的问题；而且额外引入了消息未能从内存中释放的问题。
- UnPeek-LiveData
这一部分来自 现有解决方案及各自缺陷 但是对于第二种反射 version 的方式的描述我没明白不确定说的实时性是不是对方法调用栈的判断，另外第二种方式改变了 LiveData 的职责变成了一个事件总线
对比来说 ‘Event 事件包装器’ 和 SingleLiveEvent 的问题在于它仅限于一个观察者如果添加了多个，则只会调用一个，并且不能保证哪一个明显是不太符合需求的，反射修改 version 的方式是实现了一个事件总线改变了 LiveData 的职责边界需要针对不同的需求来确定是否使用。下面来看一下 UnPeek-LiveData 的实现
public class ProtectedUnPeekLiveData<T> extends LiveData<T> {
private final static int START_VERSION = -1;
private final AtomicInteger mCurrentVersion = new AtomicInteger(START_VERSION);
protected boolean isAllowNullValue;
@Override
public void observe(@NonNull LifecycleOwner owner, @NonNull Observer<? super T> observer) {
super.observe(owner, createObserverWrapper(observer, mCurrentVersion.get()));
}
@Override
public void observeForever(@NonNull Observer<? super T> observer) {
super.observeForever(createObserverWrapper(observer, mCurrentVersion.get()));
}
public void observeSticky(@NonNull LifecycleOwner owner, @NonNull Observer<T> observer) {
super.observe(owner, createObserverWrapper(observer, START_VERSION));
}
public void observeStickyForever(@NonNull Observer<? super T> observer) {
super.observeForever(createObserverWrapper(observer, START_VERSION));
}
@Override
protected void setValue(T value) {
mCurrentVersion.getAndIncrement();
super.setValue(value);
}
class ObserverWrapper implements Observer<T> {
private final Observer<? super T> mObserver;
private int mVersion = START_VERSION;
public ObserverWrapper(@NonNull Observer<? super T> observer, int version) {
this.mObserver = observer;
this.mVersion = version;
}
@Override
public void onChanged(T t) {
if (mCurrentVersion.get() > mVersion && (t != null || isAllowNullValue)) {
mObserver.onChanged(t);
}
}
@SuppressWarnings("unchecked")
@Override
public boolean equals(Object o) {
if (this == o) {
return true;
}
if (o == null || getClass() != o.getClass()) {
return false;
}
ObserverWrapper that = (ObserverWrapper) o;
return Objects.equals(mObserver, that.mObserver);
}
@Override
public int hashCode() {
return Objects.hash(mObserver);
}
}
@Override
public void removeObserver(@NonNull Observer<? super T> observer) {
if (observer.getClass().isAssignableFrom(ObserverWrapper.class)) {
super.removeObserver(observer);
} else {
super.removeObserver(createObserverWrapper(observer, START_VERSION));
}
}
private ObserverWrapper createObserverWrapper(@NonNull Observer<? super T> observer, int version) {
return new ObserverWrapper(observer, version);
}
public void clear() {
super.setValue(null);
}
}
与第二种反射修改 version 的方式有些相似只不过没有使用反射，而是对每一个观察者进行包装自己维护了版本保证先添加的观察者才可以收到数据变化的通知，相对 LiveData 的职责来说可能是更好的处理
MediatorLiveData
MediatorLiveData 是 MutableLiveData 的子类，它将其它 LiveData 作为数据源来进行监听，也可将其作为普通的 MutableLiveData 进行使用，简单来说既可以观察其他的 LiveData 也可以作为 LiveData 被其他观察者观察
private val nameLiveData = MutableLiveData<String>()
private val nameLengthLiveData = MediatorLiveData<Int>()
// 将 nameLiveData 作为数据源
// 只要 nameLiveData 的数据发生变化 nameLengthLiveData 就能收到通知
nameLengthLiveData.addSource(nameLiveData) { name ->
nameLengthLiveData.value = name.length
}
nameLengthLiveData.observe(this, Observer {
Log.e("TAG", "name length: $it")
})
上面的例子来自参考与感谢
使用比较简单，先创建一个 MediatorLiveData 对象添加一个 MutableLiveData 作为数据源，然后添加一个自己的观察者，当数据源变化的时候调用 MediatorLiveData 的 setValue 改变 MediatorLiveData 对象的值就可以通知到它的观察者了，下面看一下源码
public class MediatorLiveData<T> extends MutableLiveData<T> {
private SafeIterableMap<LiveData<?>, Source<?>> mSources = new SafeIterableMap<>();
@MainThread
public <S> void addSource(@NonNull LiveData<S> source, @NonNull Observer<? super S> onChanged) {
// 将源 LiveData 和源 LiveData 观察者包装成一个 Source 对象
Source<S> e = new Source<>(source, onChanged);
Source<?> existing = mSources.putIfAbsent(source, e);
// 如果已经添加过同一个 LiveData 并且绑定的不是同一个观察者对象则抛出异常
if (existing != null && existing.mObserver != onChanged) {
throw new IllegalArgumentException(
"This source was already added with the different observer");
}
// 是同一个观察者对象则忽略
if (existing != null) {
return;
}
// 如果有活跃的观察者
if (hasActiveObservers()) {
// 则 "激活" Source（Source 同时也是一个 Observer 让 Source 观察传入的 LiveData 接收到数据变化后再转发给原始传入的 onChanged）
// 因为最后数据的变化是要通知到当前对象（MediatorLiveData）的观察者的
// 如果当前对象（MediatorLiveData）没有处于活跃状态的观察者也没有必要对源数据进行观察
e.plug();
}
}
@MainThread
public <S> void removeSource(@NonNull LiveData<S> toRemote) {
// 将数据源从链表中移除
Source<?> source = mSources.remove(toRemote);
if (source != null) {
// 解除 Source 对数据源 toRemote 的观察
source.unplug();
}
}
@CallSuper
@Override
protected void onActive() {
// 当有活跃状态的观察者时会调用此方法
for (Map.Entry<LiveData<?>, Source<?>> source : mSources) {
// 让 Source 观察所有的数据源
source.getValue().plug();
}
}
@CallSuper
@Override
protected void onInactive() {
// 当没有活跃状态的观察者时会调用次方法
for (Map.Entry<LiveData<?>, Source<?>> source : mSources) {
// 解除 Source 观察所有的数据源
source.getValue().unplug();
}
}
// Source 是一个静态内部类同时实现了 Observer 接口
private static class Source<V> implements Observer<V> {
// 数据源
final LiveData<V> mLiveData;
// 添加数据源（调用 addSource 方法）时一起传入的数据源的观察者
final Observer<? super V> mObserver;
// 同样维护了一个版本号
int mVersion = START_VERSION;
Source(LiveData<V> liveData, final Observer<? super V> observer) {
mLiveData = liveData;
mObserver = observer;
}
void plug() {
// 调用此方法后 Source 才会收到数据源变化的通知才会回调给传入的 mObserver
// 在传入的 mObserver 中设置当前 MediatorLiveData 改变
// 最终观察此 MediatorLiveData 的观察者才可以收到通知
// 因为最后是否通知到取决于当前 MediatorLiveData 传入的观察者是否绑定生命周期
// 所以对数据源的观察使用的 observeForever
mLiveData.observeForever(this);
}
void unplug() {
// 不再观察数据源
// 如果所有的数据源都不再观察 MediatorLiveData 的观察者也肯定收不到通知了
mLiveData.removeObserver(this);
}
@Override
public void onChanged(@Nullable V v) {
// 数据源数据变化是回调此方法
if (mVersion != mLiveData.getVersion()) {
mVersion = mLiveData.getVersion();
// 转发给添加数据源时传入的观察者
// 在这个观察者里做一些处理后设置当前 MediatorLiveData 数据改变
// 最终任意数据源发生改变后当前 MediatorLiveData 的观察者就可以收到通知了
mObserver.onChanged(v);
}
}
}
}
MediatorLiveData 的源码比较简单但是设计的我觉得还是很精妙的，既充当观察者又充当被观察者并且只有当自己有活跃中的观察者时才对数据源进行观察，MediatorLiveData 最为方便的一点就是允许通过多次调用 addSource 方法来添加多个不同的数据源，这使得我们可以将不同的数据源（例如：本地数据库缓存、网络请求结果等）进行汇总，最后再统一从一个出口进行分发
Transformations
Transformations 是一个工具类，提供了三个工具方法都是对 MediatorLiveData 的封装以简化使用，并且都只允许在主线程调用
// androidx.arch.core.util.Function
public interface Function<I, O> {
O apply(I input);
}
@MainThread
@NonNull
public static <X, Y> LiveData<Y> map(
@NonNull LiveData<X> source,
@NonNull final Function<X, Y> mapFunction) {
final MediatorLiveData<Y> result = new MediatorLiveData<>();
result.addSource(source, new Observer<X>() {
@Override
public void onChanged(@Nullable X x) {
result.setValue(mapFunction.apply(x));
}
});
return result;
}
map 方法比较简单，就是对 MediatorLiveData 添加数据源的操作进行了一个封装，把对原数据的转换抽象成了一个接口，在源数据变化的时候对数据进行转换的结果直接设置给 MediatorLiveData 观察者就可以收到通知了
@MainThread
@NonNull
public static <X, Y> LiveData<Y> switchMap(
@NonNull LiveData<X> source,
@NonNull final Function<X, LiveData<Y>> switchMapFunction) {
// 先创建 MediatorLiveData 对象
final MediatorLiveData<Y> result = new MediatorLiveData<>();
// 添加数据源
result.addSource(source, new Observer<X>() {
// 记录上次调用传入的函数的返回值
LiveData<Y> mSource;
@Override
public void onChanged(@Nullable X x) {
// 与 map 的区别是 switchMap 传入的 switchMapFunction 的返回值是 LiveData
// switchMapFunction 的调用是惰性的当传入的 source 数据变化时才会被调用
LiveData<Y> newLiveData = switchMapFunction.apply(x);
// 如果结果没有改变不必再次添加为数据源
if (mSource == newLiveData) {
return;
}
// 如果上次的数据源不为空先解除对其的观察
if (mSource != null) {
result.removeSource(mSource);
}
// 记录数据源
mSource = newLiveData;
if (mSource != null) {
// 添加数据源
result.addSource(mSource, new Observer<Y>() {
@Override
public void onChanged(@Nullable Y y) {
// 调用 MediatorLiveData 对象的 setValue 方法通知观察者
result.setValue(y);
}
});
}
}
});
// 返回 MediatorLiveData 对象
return result;
}
先将传入的 LiveData 添加为数据源，当数据源变化时收到通知后才会触发执行传入的函数，函数的返回值还是一个 LiveData 将其添加为数据源，并且多次调用会先移除观察之前的 LiveData 在 LiveData 数据变化时调用 MediatorLiveData 的 setValue 通知观察者
@MainThread
@NonNull
public static <X> LiveData<X> distinctUntilChanged(@NonNull LiveData<X> source) {
// 创建 MediatorLiveData 对象
final MediatorLiveData<X> outputLiveData = new MediatorLiveData<>();
// 添加数据源
outputLiveData.addSource(source, new Observer<X>() {
// 标记是首次
boolean mFirstTime = true;
@Override
public void onChanged(X currentValue) {
// 数据源变化时触发
final X previousValue = outputLiveData.getValue();
// 首次
// 上一次的值为 null 新值不为 null
// 上一次的值不为 null 并且与新值不相等
if (mFirstTime
|| (previousValue == null && currentValue != null)
|| (previousValue != null && !previousValue.equals(currentValue))) {
mFirstTime = false;
// 设置新值
outputLiveData.setValue(currentValue);
}
}
});
return outputLiveData;
}
distinctUntilChanged 的作用是去重比较简单
总结
- LiveData 可感知生命周期只更新处于活跃（STARTED 或 RESUMED）状态的观察则并且当观察者绑定的组件变为 DESTROYED 状态时自动移除（针对 observe 对于 observeForever 需要手动移除 ）观察者避免可能 NPE 和内存泄漏
- 同一个观察者不能绑定多个生命周期组件（实现了 LifecycleOwner 接口）
- 同一个观察者不能同时调用 observe 和 observeForever
- 连续调用多次 postValue 中间值可能会丢失会出现部分观察者收到中间值所有观察者收到最后的值的情况
- 当观察者所绑定的生命周期组件处于非活跃状态时多次更新 LiveData 变为活跃状态时只会收到最后一次的值
- LiveData 是粘性的，这是一把双刃剑
- 当 LiveData 的活跃观察者数量从 0 变为 1 时会回调 onInactive 当活跃观察者数量从 1 变为 0 时会回调 onActive 开发者可以根据需要做处理
- MediatorLiveData 可以将 LiveData 作为数据源并且可以作为普通 LiveData 使用
- MediatorLiveData 只有当有活跃状态的观察者时才会观察源 LiveData
参考与感谢
Android消息总线的演进之路：用LiveDataBus替代RxBus、EventBus
从源码看 Jetpack（3）- LiveData 源码详解
从源码看 Jetpack（4）- LiveData 衍生物源码详解
Jetpack LiveData 的设计理念及改进
UnPeek-LiveData
Jetpack篇——LiveData扩展之MediatorLiveData源码分析
