LiveData 源码分析之事件总线 LiveBus 实现
Posted code小生
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了LiveData 源码分析之事件总线 LiveBus 实现相关的知识，希望对你有一定的参考价值。
code小生,一个专注 android 领域的技术平台
前言
最近的项目重构中加入 LiveData 框架，并且小码的 T-MVVM 架构项目中也是用了LiveData 框架，好不好用你试试就知道（小码口头禅），于是乎翻了翻 LiveData源代码，看了看其实现过程，在此做下笔记。
什么是 LiveData
LiveData是一个数据持有类。它具有以下特点：
数据可以被观察者订阅；
能够感知组件（Fragment、Activity、Service）的生命周期；
只有在组件出于激活状态才会通知观察者有数据更新；
LiveData能为我们做什么
能够保证数据和UI统一,LiveData采用了观察者模式，LiveData是被观察者，当数据有变化时会通知UI。
减少内存泄漏,LiveData能够感知到组件的生命周期，当组件处于DESTROYED状态时，观察者对象会被清除,当Activity停止时不会导致Crash,因为组件处于非激活状态时，不会收到LiveData中数据变化的通知。
不需要额外的手动处理来响应生命周期的变化，因为LiveData能够感知组件的生命周期，所以就完全不需要在代码中告诉LiveData组件的生命周期状态。
组件和数据相关的内容能实时更新,组件在前台的时候能够实时收到数据改变的通知，当组件从后台到前台来时，LiveData能够将最新的数据通知组件，因此保证了组件中和数据相关的内容能够实时更新。
若果横竖屏切换（configuration change）时，不需要额外的处理来保存数据,当屏幕方向变化时，组件会被recreate，然而系统并不能保证你的数据能够被恢复的。当我们采用LiveData保存数据时，因为数据和组件分离了。当组件被recreate，数据还是存在LiveData中，并不会被销毁。资源共享.
接下来先从使用玩起，LiveData的2种创建方式。
直接使用MutableLiveData对象，
自行继承LiveData类，
发起数据通知2种方式：postValue和setValue：
private MutableLiveData<String> mData=new MutableLiveData();
mData.postValue("Hello LiveData");
// mData.setValue("Hello LiveData");
此处postValue和setValue有什么区别呢？，下文会提，
注册观察者并监听数据变化：
mData.observe(this, new Observer<String>() {
@Override
public void onChanged(@Nullable String str) {
Toast.makeText(this, str, Toast.LENGTH_SHORT).show();
}
});
使用非常简单，仅此而已
通过observe方法注册观察者，哪咋们就看看observe里面具体干了什么,源代码165行走起，
//注册观察
@MainThread
public void observe(@NonNull LifecycleOwner owner, @NonNull Observer<T> observer) {
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);
}
observe 方法接收2个参数，一个是具有生命周期的 LifecycleOwner，另一个是观察者 Observer
class LifecycleBoundObserver extends ObserverWrapper implements GenericLifecycleObserver {
@NonNull final LifecycleOwner mOwner;
LifecycleBoundObserver(@NonNull LifecycleOwner owner, Observer<T> observer) {
super(observer);
mOwner = owner;
}
@Override
boolean shouldBeActive() {
return mOwner.getLifecycle().getCurrentState().isAtLeast(STARTED);
}
@Override
public void onStateChanged(LifecycleOwner source, Lifecycle.Event event) {
if (mOwner.getLifecycle().getCurrentState() == DESTROYED) {
removeObserver(mObserver);
return;
}
activeStateChanged(shouldBeActive());
}
@Override
boolean isAttachedTo(LifecycleOwner owner) {
return mOwner == owner;
}
@Override
void detachObserver() {
mOwner.getLifecycle().removeObserver(this);
}
}
此方法中，我们先看 onStateChanged()方法，当生命周期变化时会回调，如果getCurrentState() == DESTROYED 则 removeObserver，反之则调用父类ObserverWrapper 的 activeStateChanged（）方法，源代码382行走起，
private abstract class ObserverWrapper {
final Observer<T> mObserver;
boolean mActive;
int mLastVersion = START_VERSION;
ObserverWrapper(Observer<T> observer) {
mObserver = observer;
}
abstract boolean shouldBeActive();
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;
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);
}
}
}
activeStateChanged() 是干嘛的，首先判断 activeState 新旧状态是否相同，不同则把新的状态赋给 mActive，是生命周期状态处于 ACTIVE 情况下的逻辑处理。如果新的状态和旧的状态相同则直接返回。这里有个常量LiveData.this.mActiveCount，看注释可以理解为观察者处于活动状态
个数，往下看 if (wasInactive && mActive) 如果 mActiveCount=0 并且mActive 为 true,即观察者处于活动状态
个数从0变为1个则调用onActive(); 观察者处于活动状态
个数从1变为0时则调用onInactive()。然而onActive()，onInactive()并没有任何实现代码。好了，接下来继续往下看dispatchingValue(this);应该就是数据变化消息调度。源代码112行走起，
private 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<T>, ObserverWrapper>> iterator =
mObservers.iteratorWithAdditions(); iterator.hasNext(); ) {
considerNotify(iterator.next().getValue());
if (mDispatchInvalidated) {
break;
}
}
}
} while (mDispatchInvalidated);
mDispatchingValue = false;
}
前面的几行if 判断姑且先不看，先看从if(initiator != null)开始看，如果initiator!= null调用considerNotify(initiator)方法;源代码91行走起，
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;
//noinspection unchecked
observer.mObserver.onChanged((T) mData);
}
看见没有最后一行代码 observer.mObserver.onChanged((T) mData); Observer的数据变化回调；好了我们再回过头看看initiator == null的逻辑，
for (Iterator<Map.Entry<Observer<T>, ObserverWrapper>> iterator =
mObservers.iteratorWithAdditions(); iterator.hasNext(); ) {
considerNotify(iterator.next().getValue());
if (mDispatchInvalidated) {
break;
}
}
如果 initiator == null 则会通过迭代器 mObservers 遍历获取ObserverWrapper，最终还是调用 considerNotify 方法；既然有取ObserverWrapper，咋们再看看在哪儿存的，在源码171行：
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);
mObservers.putIfAbsent(observer, wrapper)存入容器中，mObservers.putIfAbsent这个添加数据的方式貌似很少见，于是乎在看看mObservers是个什么数据容器，成员变量中：
private SafeIterableMap<Observer<T>, ObserverWrapper> mObservers =
new SafeIterableMap<>();
这是个什么鬼，貌似以前很少见，查阅资料发现：
SafeIterableMap有以下特性：
1;支持键值对存储，用链表实现，模拟成Map的接口
2:支持在遍历的过程中删除任意元素，不会触发ConcurrentModifiedException
3:非线程安全
感兴趣的可自行查阅，此处不再详细介绍，
最后 addObserver 添加注册。
下面在看看数据发起通知的逻辑，数据发起通知有2中方式：
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);
}
}
setValue 和 postValue2 中方式，我们先看看 setValue 代码：
@MainThread
protected void setValue(T value) {
assertMainThread("setValue");
mVersion++;
mData = value;
dispatchingValue(null);
}
private static void assertMainThread(String methodName) {
if (!ArchTaskExecutor.getInstance().isMainThread()) {
throw new IllegalStateException("Cannot invoke " + methodName + " on a background"
+ " thread");
}
}
setValue 第一行代码 assertMainThread("setValue");则是判断是否在主线程，所以貌似 setValue 方式必须在主线程中执行，如果非主线程则抛出异常。
再看看 postValue:
protected void postValue(T value) {
boolean postTask;
synchronized (mDataLock) {
postTask = mPendingData == NOT_SET;
mPendingData = value;
}
if (!postTask) {
return;
}
ArchTaskExecutor.getInstance().postToMainThread(mPostValueRunnable);
}
则 postValue 调用 postToMainThread 方法，最终还是用过 setValue 方式：
private final Runnable mPostValueRunnable = new Runnable() {
@Override
public void run() {
Object newValue;
synchronized (mDataLock) {
newValue = mPendingData;
mPendingData = NOT_SET;
}
//noinspection unchecked
setValue((T) newValue);
}
};
因此最终明白为什么 setValue 方法只能在主线程中调用，postValue可以在任何线程中调用，如果是在后台子线程中更新 LiveData 的值，必须调用 postValue。
至此大概明白了LiveData是干嘛的 ，怎么干的，总的来说还是很好用的，于是乎就有个小小的想法，平时用Eventbus，RxBus什么的，貌似感觉LiveData也可以实现事件通信，既然有了想法，那就干呗，下面就开始编写LiveBus。
/**
* 事件总线
*
* @author：tqzhang on 18/9/11 17:22
*/
public class LiveBus {
private static volatile LiveBus instance;
private final Map<Object, MutableLiveData<Object>> mLiveBus;
private LiveBus() {
mLiveBus = new HashMap<>();
}
public static LiveBus getDefault() {
if (instance == null) {
synchronized (LiveBus.class) {
if (instance == null) {
instance = new LiveBus();
}
}
}
return instance;
}
public <T> MutableLiveData<T> subscribe(Object eventKey) {
checkNotNull(eventKey);
return (MutableLiveData<T>) subscribe(eventKey, Object.class);
}
public <T> MutableLiveData<T> subscribe(Object eventKey, Class<T> tMutableLiveData) {
checkNotNull(eventKey);
checkNotNull(tMutableLiveData);
if (!mLiveBus.containsKey(eventKey)) {
mLiveBus.put(eventKey, new MutableLiveData<>(true));
} else {
MutableLiveData liveBusData = mLiveBus.get(eventKey);
liveBusData.isFirstSubscribe = false;
}
return (MutableLiveData<T>) mLiveBus.get(eventKey);
}
public <T> MutableLiveData<T> postEvent(Object eventKey, T value) {
checkNotNull(eventKey);
MutableLiveData<T> mutableLiveData = subscribe(eventKey);
mutableLiveData.postValue(value);
return mutableLiveData;
}
}
代码相对比较简单，看似没有什么问题。行不行实践出真知，跑跑看，基本功能貌似实现了，然而在同一个页面多次注册相同事件名的事件事，会多走一次onChanged()回调方法，这是为什么呢？我们继续在源代码看，在considerNotify()方法里的104行这里有一个if判断。
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;
//noinspection unchecked
observer.mObserver.onChanged((T) mData);
}
如果 ObserverWrapper 的 mLastVersion 小于 LiveData 的 mVersion，就会走回调 mObserver 的 onChanged 方法，LiveData 每次 setValue 或者postValue 都会使其 version 加 1 然而每个订阅者，其默认version都是-1，当LiveData 设置这个 ObserverWrapper 的时候，如果 LiveData 的 version 大于 ObserverWrapper 的 version，LiveData 就会把当前 version 赋给Observer。解决这个问题就是要把 ObserverWrapper 的 version 设置成跟LiveData 的 version 一致就 OK 了。下面的解决办法是过滤第一次注册时候的回调方法。
/**
* 事件总线
*
* @author：tqzhang on 18/9/11 17:22
*/
public class LiveBus {
private static volatile LiveBus instance;
private final Map<Object, MutableLiveData<Object>> mLiveBus;
private LiveBus() {
mLiveBus = new HashMap<>();
}
public static LiveBus getDefault() {
if (instance == null) {
synchronized (LiveBus.class) {
if (instance == null) {
instance = new LiveBus();
}
}
}
return instance;
}
public <T> MutableLiveData<T> subscribe(Object eventKey) {
checkNotNull(eventKey);
return (MutableLiveData<T>) subscribe(eventKey, Object.class);
}
public <T> MutableLiveData<T> subscribe(Object eventKey, Class<T> tMutableLiveData) {
checkNotNull(eventKey);
checkNotNull(tMutableLiveData);
if (!mLiveBus.containsKey(eventKey)) {
mLiveBus.put(eventKey, new LiveBusData<>(true));
} else {
LiveBusData liveBusData = mLiveBus.get(eventKey);
liveBusData.isFirstSubscribe = false;
}
return (MutableLiveData<T>) mLiveBus.get(eventKey);
}
public <T> MutableLiveData<T> postEvent(Object eventKey, T value) {
checkNotNull(eventKey);
MutableLiveData<T> mutableLiveData = subscribe(eventKey);
mutableLiveData.postValue(value);
return mutableLiveData;
}
public static class LiveBusData<T> extends MutableLiveData<T> {
private boolean isFirstSubscribe;
public LiveBusData(boolean isFirstSubscribe) {
this.isFirstSubscribe = isFirstSubscribe;
}
@Override
public void observe(@NonNull LifecycleOwner owner, @NonNull Observer<T> observer) {
super.observe(owner, new ObserverWrapper<>(observer, isFirstSubscribe));
}
}
private static class ObserverWrapper<T> implements Observer<T> {
private Observer<T> observer;
private boolean isChanged;
private ObserverWrapper(Observer<T> observer,boolean isFirstSubscribe) {
this.observer = observer;
isChanged = isFirstSubscribe;
}
@Override
public void onChanged(@Nullable T t) {
if (isChanged) {
if (observer != null) {
observer.onChanged(t);
}
} else {
isChanged = true;
}
}
}
通过变量过滤掉第一次订阅的回调，简单的几十行代码实现类似Eventbus，RxBus的功能，小码亲测了，挺好使的。
//发起订阅事件
LiveBus.getDefault().postEvent("LiveData","hello LiveData");
//观察订阅
LiveBus.getDefault().subscribe("LiveData").observe(this, new Observer<String>() {
@Override
public void onChanged(@Nullable String s) {
Toast.makeText(activity, s, Toast.LENGTH_SHORT).show();
}
});
源码
LiveBus 完整代码以及实战项目：
https://github.com/SelfZhangTQ/T-MVVM
欢迎大家交流，由于小码水平有限，有意见或者是建议请各位指正。
分享技术我是认真的
以上是关于LiveData 源码分析之事件总线 LiveBus 实现的主要内容，如果未能解决你的问题，请参考以下文章