前言：
里面有三个类：
- LiveData ：抽象类
- MutableLiveData： LiveData的实现类
- Observer ： 观察者接口
一、发生事件
public MutableLiveData<String> liveData = new MutableLiveData<>();
//在任意线程中
liveData .postValue("SSS");
//在主线程中
liveData .setValue("SSS");
- 1）setValue 过程
private volatile Object mData;
@MainThread
protected void setValue(T value) {
assertMainThread("setValue");
//mVersion 表示数据发生了变化
mVersion++;
// 保存了这次变化的数据
mData = value;
//分发数据变化，调用回调函数
dispatchingValue(null);
}
- 2）postValue过程
static final Object NOT_SET = new Object();
volatile Object mPendingData = NOT_SET;
protected void postValue(T value) {
boolean postTask;
synchronized (mDataLock) {
postTask = mPendingData == NOT_SET;
mPendingData = value;
}
if (!postTask) {
return;
}
// postValue 可以从后台线程调用，因为它会在主线程中执行任务。内部通过 Handler 切换的
//执行 DefaultTaskExecutor#postToMainThread方法用Handler切换主线程
ArchTaskExecutor.getInstance().postToMainThread(mPostValueRunnable);
}
会切换到主线程后，在调用 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);
}
};
void dispatchingValue(@Nullable ObserverWrapper initiator) {
if (mDispatchingValue) {
mDispatchInvalidated = true;
return;
}
//进入while 循环前，设置为true，如果此时另外一个数据发生变化，到了此函数中就直接在上面返回了
mDispatchingValue = true;
do {
//开始for循环前，设置为false，for循环完，也会退出while循环
mDispatchInvalidated = false;
if (initiator != null) {
//给订阅用 observe
considerNotify(initiator);
initiator = null;
} else {
//给发送事件用，setValue/postValue
for (Iterator<Map.Entry<Observer<? super T>, ObserverWrapper>> iterator =
mObservers.iteratorWithAdditions(); iterator.hasNext(); ) {
considerNotify(iterator.next().getValue());
//这里mDispatchInvalidated 为true，表示在while循环未结束的时候，有其他数据发生变化，并调用了该函数
//在上面的if判断中设置了 mDispatchInvalidated = true，
// 结束本次for循环，没有退出while循环，开始下一次for循环
if (mDispatchInvalidated) {
break;
}
}
}
} while (mDispatchInvalidated);
//退出while 后，设置为false，正常处理数据变化
mDispatchingValue = false;
}
- 3）、线程切换过程
ArchTaskExecutor.getInstance()在ArchTaskExecutor类中通过单例创建的。
public static ArchTaskExecutor getInstance() {
if (sInstance != null) {
return sInstance;
}
synchronized (ArchTaskExecutor.class) {
if (sInstance == null) {
sInstance = new ArchTaskExecutor();
}
}
return sInstance;
}
private ArchTaskExecutor() {
//创建 DefaultTaskExecutor 在里面执行切换线程
mDefaultTaskExecutor = new DefaultTaskExecutor();
mDelegate = mDefaultTaskExecutor;
}
我们再看xx..postToMainThread(mPostValueRunnable)的执行，实际实在DefaultTaskExecutor中执行的。
@Nullable
private volatile Handler mMainHandler;
@Override
public void executeOnDiskIO(Runnable runnable) {
mDiskIO.execute(runnable);
}
@Override
public void postToMainThread(Runnable runnable) {
if (mMainHandler == null) {
synchronized (mLock) {
if (mMainHandler == null) {
mMainHandler = createAsync(Looper.getMainLooper());
}
}
}
//用 post 发送消息
mMainHandler.post(runnable);
}
//创建 Handler
private static Handler createAsync(@NonNull Looper looper) {
if (Build.VERSION.SDK_INT >= 28) {
return Handler.createAsync(looper);
}
if (Build.VERSION.SDK_INT >= 16) {
try {
return Handler.class.getDeclaredConstructor(Looper.class, Handler.Callback.class,
boolean.class)
.newInstance(looper, null, true);
} catch (IllegalAccessException ignored) {
} catch (InstantiationException ignored) {
} catch (NoSuchMethodException ignored) {
} catch (InvocationTargetException e) {
return new Handler(looper);
}
}
return new Handler(looper);
}
二、注册观察者
private SafeIterableMap<Observer<? super T>, ObserverWrapper> mObservers = new SafeIterableMap<>();
@MainThread
public void observe(@NonNull LifecycleOwner owner, @NonNull Observer<? super T> observer) {
assertMainThread("observe");
//如果生命周期销毁，直接返回
if (owner.getLifecycle().getCurrentState() == DESTROYED) {
// ignore
return;
}
//把观察者和LifecycleObserver进行绑定，监听监听生命周期的变化
LifecycleBoundObserver wrapper = new LifecycleBoundObserver(owner, observer);
//observer作为key，LifecycleBoundObserver 作为value保存在 SafeIterableMap中
//把observer 保存在 mObservers
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;
}
//绑定Lifecycle的Observer事件
owner.getLifecycle().addObserver(wrapper);
}
在Lifecycle的LiveData#LifecycleBoundObserver#onStateChanged中回调事件
@Override
public void onStateChanged(@NonNull LifecycleOwner source,
@NonNull Lifecycle.Event event) {
//如果生命周期销毁，移除观察者
if (mOwner.getLifecycle().getCurrentState() == DESTROYED) {
removeObserver(mObserver);
return;
}
//处理观察者事件
activeStateChanged(shouldBeActive());
}
调用LiveData的内部类ObserverWrapper的activeStateChanged方法
void activeStateChanged(boolean newActive) {
// 当前的生命周期和上一次的生命周期状态，是否发生变化，没有发生变化，就直接返回。
// onStart-onPause 为 true 在这之外的生命周期为false
if (newActive == mActive) {
return;
}
//mActive 标记 activity 是否是活跃状态
// 当 STARTED 、RESUMED状态为 true, 反之为 false
mActive = newActive;
boolean wasInactive = LiveData.this.mActiveCount == 0;
// 当 mActive = true 时，mActiveCount 活跃的 Activity 个数加一，否则减一
LiveData.this.mActiveCount += mActive ? 1 : -1;
if (wasInactive && mActive) {
// 这是一个空函数，活跃态
onActive();
}
if (LiveData.this.mActiveCount == 0 && !mActive) {
// 这是一个空函数，非活跃态
onInactive();
}
if (mActive) {
//生命周期状态从Inactive 到 Active， 就会调用回调函数
dispatchingValue(this);
}
}
void dispatchingValue(@Nullable ObserverWrapper initiator) {
if (mDispatchingValue) {
mDispatchInvalidated = true;
return;
}
//进入while 循环前，设置为true，如果此时另外一个数据发生变化，到了此函数中就直接在上面返回了
mDispatchingValue = true;
do {
//开始for循环前，设置为false，for循环完，也会退出while循环
mDispatchInvalidated = false;
if (initiator != null) {
//给注册用
considerNotify(initiator);
initiator = null;
} else {
//给发送事件用，setValue/postValue
for (Iterator<Map.Entry<Observer<? super T>, ObserverWrapper>> iterator =
mObservers.iteratorWithAdditions(); iterator.hasNext(); ) {
considerNotify(iterator.next().getValue());
//这里mDispatchInvalidated 为true，表示在while循环未结束的时候，有其他数据发生变化，并调用了该函数
//在上面的if判断中设置了 mDispatchInvalidated = true，
// 结束本次for循环，没有退出while循环，开始下一次for循环
if (mDispatchInvalidated) {
break;
}
}
}
} while (mDispatchInvalidated);
//退出while 后，设置为false，正常处理数据变化
mDispatchingValue = false;
}
这里有两个变量
- mDispatchingValue 这个变量用来控制，是否进入while 循环，以及while 循环 是否已经结束
- mDispatchInvalidated 这个变量用来控制for 循环是否要重新开始
private void considerNotify(ObserverWrapper observer) {
if (!observer.mActive) {
return;
}
// 如果当前的生命周期是非活跃，就不回调onChanged函数，
// 在LifecycleBoundObserver 中记录状态，当生命周期变为活跃，就回去更新数据
if (!observer.shouldBeActive()) {
observer.activeStateChanged(false);
return;
}
//数据发生变化了
if (observer.mLastVersion >= mVersion) {
return;
}
observer.mLastVersion = mVersion;
//调用Observer的onChanged方法返回数据
//mData 就是改变的事件
observer.mObserver.onChanged((T) mData);
}
三、移除观察者
@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);
}
四、重写 MutableLiveData做到只有一个观察者可以接到通知改变
LiveData是一个一对多的关系。
其中的机制是用AtomicBoolean记录setValue状态， 只有当setValue被调用后，才会发送事件。发送完毕后设回false。这样阻止了注册时发送数据。
public class SingleLiveEvent<T> extends MutableLiveData<T> {
private static final String TAG = "SingleLiveEvent";
private final AtomicBoolean mPending = new AtomicBoolean(false);
@MainThread
public void observe(LifecycleOwner owner, final Observer<? super T> observer) {
if (hasActiveObservers()) {
Log.w(TAG, "Multiple observers registered but only one will be notified of changes.");
}
// Observe the internal MutableLiveData
super.observe(owner, new Observer<T>() {
@Override
public void onChanged(@Nullable T t) {
if (mPending.compareAndSet(true, false)) {
observer.onChanged(t);
}
}
});
}
@MainThread
public void setValue(@Nullable T t) {
mPending.set(true);
super.setValue(t);
}
/**
* Used for cases where T is Void, to make calls cleaner.
*/
@MainThread
public void call() {
setValue(null);
}
}
五、解决粘性事件
使用hook version的方法。这是没有办法的办法。但是功能适配性却最好。没有了粘性事件问题。这也是LiveDataBus源码中的方式。
原理是利用LiveData的mVersion和ObserverWrapper的mLastVersion之间的关系。hook的目的是跳过mVersion和mLastVersion比较这一步，使得onChanged可以执行!
初始注册时mLastVersion是-1, 而LiveData如果之前已经设过值，则mVersion大于-1。所以 mVersion> mLastVersion导致了粘性事件。
private void considerNotify(ObserverWrapper observer) {
if (!observer.mActive) {
return;
}
if (!observer.shouldBeActive()) {
observer.activeStateChanged(false);
return;
}
//hook的目的是跳过这一步，使得onChanged可以执行!
if (observer.mLastVersion >= mVersion) {
return;
}
observer.mLastVersion = mVersion;
//noinspection unchecked
observer.mObserver.onChanged((T) mData);
public 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);
}
网友评论