1. LiveDataBus 的封装
- 通过 map 维护一个消息事件和 MutableLiveData 的映射关系，MutableLiveData 的类型默认为 Object，接收任意类型，实现总线通信
- 将 LiveDataBus 封装为一个单例类。
- 消息注册时，如果当前 map 中不存在，则先将消息和对应的 MutableLiveData 对象放入维护的 map 中，添加映射关系，返回当前 map 中缓存的 MutableLiveData 对象
1.1 LiveDataBus 的组成
- 消息 消息可以是任何的Object，可以定义不同类型的消息，如 Boolean、String。也可以定义自定义类型的消息。
- 消息总线 消息总线通过单例实现，不同的消息通道存放在一个 HashMap中。
- 订阅 订阅者通过 with 获取消息通道，然后调用 observe 订阅这个通道的消息。
- 发布 发布者通过 with 获取消息通道，然后调用 setValue 或者 postValue 发布消息。
LiveData 详细解析参考 LiveData 组件使用及解析
2. 粘性消息问题解决
具体现象：当前 Activity 给未启动的 Activity 发送一个消息，Activity 在启动时能收到之前发送的消息。
3. 为什么使用 LiveData 构建数据通信总线 LiveDataBus？
3.1 使用 LiveData 的理由
- LiveData 具有的这种可观察性和生命周期感知的能力，使其非常适合作为 Android 通信总线的基础构件
- 使用者不用显示调用反注册方法。
由于 LiveData 具有生命周期感知能力，所以 LiveDataBus 只需要调用注册回调方法，而不需要显示的调用反注册方法。这样带来的好处不仅可以编写更少的代码，而且可以完全杜绝其他通信总线类框架（如EventBus、RxBus）忘记调用反注册所带来的内存泄漏的风险。
3.2 为什么要用 LiveDataBus 替代 EventBus 和 RxBus
- LiveDataBus 的实现及其简单 相对 EventBus 复杂的实现，LiveDataBus 只需要一个类就可以实现。
- LiveDataBus 可以减小 APK 包的大小 由于 LiveDataBus 只依赖 Android 官方Android Architecture Components 组件的 LiveData，没有其他依赖，本身实现只有一个类。作为比较，EventBus JAR 包大小为57kb，RxBus 依赖 RxJava 和RxAndroid，其中 RxJava2 包大小 2.2MB，RxJava1 包大小 1.1MB，RxAndroid包大小9kb。使用 LiveDataBus 可以大大减小 APK 包的大小。
- LiveDataBus 依赖方支持更好 LiveDataBus 只依赖 Android 官方 Android Architecture Components 组件的 LiveData，相比 RxBus 依赖的 RxJava 和RxAndroid，依赖方支持更好。
- LiveDataBus 具有生命周期感知 LiveDataBus 具有生命周期感知，在 Android 系统中使用调用者不需要调用反注册，相比 EventBus 和 RxBus 使用更为方便，并且没有内存泄漏风险。
4. LiveDataBus 的实现
4.1 实现一
public final class LiveDataBus {
private final Map<String, MutableLiveData<Object>> bus;
private LiveDataBus() {
bus = new HashMap<>();
}
private static class SingletonHolder {
private static final LiveDataBus DATA_BUS = new LiveDataBus();
}
public static LiveDataBus get() {
return SingletonHolder.DATA_BUS;
}
public <T> MutableLiveData<T> with(String target, Class<T> type) {
if (!bus.containsKey(target)) {
bus.put(target, new MutableLiveData<>());
}
return (MutableLiveData<T>) bus.get(target);
}
public MutableLiveData<Object> with(String target) {
return with(target, Object.class);
}
}
上述代码就实现了一个通信总线的全部功能，并且还具有生命周期感知功能，并且使用起来也及其简单：
注册订阅：
LiveDataBus.get().with("key_test", Boolean.class)
.observe(this, new Observer<Boolean>() {
@Override
public void onChanged(@Nullable Boolean aBoolean) {
}
});
发送消息：
LiveDataBus.get().with("key_test").setValue(true);
发送了一个名为”key_test”，值为 true 的事件。这个时候订阅者就会收到消息，并作相应的处理。
4.1.1 问题
在使用实现一这个 LiveDataBus 的过程中，订阅者会收到订阅之前发布的消息。无论 EventBus 或者 RxBus，订阅方都不会收到订阅之前发出的消息。对于一个消息总线，LiveDataBus 必须要解决这个问题
4.1.2 问题分析
当 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 每注册一个新的订阅者，这个订阅者立刻会收到一个回调，即使这个设置的动作发生在订阅之前。
private void considerNotify(ObserverWrapper observer) {
if (!observer.mActive) {
return;
}
// 观察者没有处于活跃状态则不分发。
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);
}
核心原因： 对于 LiveData，其初始的 version 是-1，当我们调用了其 setValue或者 postValue，其 vesion 会+1；对于每一个观察者的封装 ObserverWrapper，其初始 version 也为-1，也就是说，每一个新注册的观察者，其 version 为-1；当LiveData 设置这个 ObserverWrapper 的时候，如果 LiveData 的 version 大于 ObserverWrapper 的 version，LiveData 就会强制把当前 value 推送给 Observer。
private int mVersion = START_VERSION;
protected void setValue(T value) {
assertMainThread("setValue");
mVersion++;
mData = value;
dispatchingValue(null);
}
4.1.3 如何解决这个问题
只需要在注册一个新的订阅者的时候把 ObserverWrapper 的 version 设置成跟 LiveData 的 version 一致即可。
在 LiveData 的 observe 方法中，会创建一个 LifecycleBoundObserver，LifecycleBoundObserver 是 ObserverWrapper 的派生类。然后把这个LifecycleBoundObserver 放入一个私有 Map 容器 mObservers 中。无论ObserverWrapper 还是 LifecycleBoundObserver 都是私有的或者包可见的，所以无法通过继承的方式更改 LifecycleBoundObserver 的 version。
从 Map 容器 mObservers 中取到 LifecycleBoundObserver，然后再更改 version，通过查看 SafeIterableMap 的源码我们发现有一个 protected 的 get 方法。因此，在调用 observe 的时候，我们可以通过反射拿到 LifecycleBoundObserver，再把 LifecycleBoundObserver 的 version 设置成和 LiveData 一致即可。
对非生命周期感知的 observeForever 方法，生成的 wrapper 不是 LifecycleBoundObserver 而是 AlwaysActiveObserver，没有办法在 observeForever 调用完后再改 AlwaysActiveObserver 的 version，因为注册监听时直接调了 wrapper.activeStateChanged(true) 而不是在 LifeCircleOwner 的状态变化时。
那么对于 observeForever，如何解决这个问题呢？
既然是在调用内回调的，那么我们可以写一个 ObserverWrapper，包装真正的回调传给 observeForever。把 ObserverWrapper 传给 observeForever，回调时检查调用栈，如果回调是 observeForever 方法引起的，那么就不回调真正的订阅者。
// 包装类包裹真正的Observer，处理非生命周期感知的注册监听
private static class ObserverWrapper<T> implements Observer<T> {
private Observer<T> observer;
public ObserverWrapper(Observer<T> observer) {
this.observer = observer;
}
public void onChanged(@Nullable T t) {
if (observer != null) {
// 目标方法不调onChanged
if (isCallOnObserve()) {
return;
}
observer.onChanged(t);
}
}
private boolean isCallOnObserve() {
StackTraceElement[] stackTrace = Thread.currentThread().getStackTrace();
if (stackTrace != null && stackTrace.length > 0) {
for (StackTraceElement element : stackTrace) {
// 如果当前是LiveData对象且为observeForever方法
if ("android.arch.lifecycle.LiveData".equals(element.getClassName()) &&
"observeForever".equals(element.getMethodName())) {
return true;
}
}
}
return false;
}
}
Hook
在事件传递过程中拦截并监控事件的传输，修改事件传递流程
只要调用 setValue 版本号 mVersion 就会加1，此时版本号已经不一致导致 onChange 的调用，触发粘性事件，如果将 mObservers.observer.mLastVersion 修改为 mVersion 当前版本，就会在 mObservers.observer.onChange 调用前，也就是数据变化通知前 return 结束，这样就不调 onChange 方法 mObservers 是 Map 对象，Map 的 item 是键值对，observer 是键值对的 value，反射 Map 获取到 Entry 并获取到 value 也就是 observer
继承 MutableLiveData，重写 observe 方法，在注册监听时进行 hook 逻辑。
LiveDataBus最终实现
/**
* @desciption : 利用LivaData来实现数据传递,可用来替换EventBus
*/
public 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<? super T> observer) {
super.observe(owner, observer);
try {
hook(observer);
} catch (Exception e) {
e.printStackTrace();
}
}
// 非生命周期感知的注册监听处理，去除粘性事件
@Override
public void observeForever(@NonNull Observer<? super T> observer) {
if (!observerMap.containsKey(observer)) {
observerMap.put(observer, new ObserverWrapper(observer));
}
super.observeForever(observerMap.get(observer));
}
// 非生命周期感知取消注册监听
@Override
public void removeObserver(@NonNull Observer<? super T> observer) {
Observer realObserver = null;
if (observerMap.containsKey(observer)) {
realObserver = observerMap.remove(observer);
} else {
realObserver = observer;
}
super.removeObserver(realObserver);
}
// 去除粘性事件
private void hook(@NonNull Observer<? super T> observer) throws Exception {
//get wrapper's version
//1.得到mLastVersion
//获取到LivData的类中的mObservers对象
Class<LiveData> classLiveData = LiveData.class;
Field fieldObservers = classLiveData.getDeclaredField("mObservers");
fieldObservers.setAccessible(true);
//获取到这个成员变量的对象
Object objectObservers = fieldObservers.get(this);
//得到map对象的class对象
Class<?> classObservers = objectObservers.getClass();
//获取到mObservers对象的get方法
Method methodGet = classObservers.getDeclaredMethod("get", Object.class);
methodGet.setAccessible(true);
//执行get方法
Object objectWrapperEntry = methodGet.invoke(objectObservers, observer);
//取到entry中的value
Object objectWrapper = null;
if (objectWrapperEntry instanceof Map.Entry) {
objectWrapper = ((Map.Entry) objectWrapperEntry).getValue();
}
if (objectWrapper == null) {
throw new NullPointerException("Wrapper can not be bull!");
}
//得到observerWraperr的类对象
Class<?> classObserverWrapper = objectWrapper.getClass().getSuperclass();
Field fieldLastVersion = classObserverWrapper.getDeclaredField("mLastVersion");
fieldLastVersion.setAccessible(true);
//get livedata's version
//2.得到mVersion
Field fieldVersion = classLiveData.getDeclaredField("mVersion");
fieldVersion.setAccessible(true);
//3.mLastVersion=mVersion
Object objectVersion = fieldVersion.get(this);
//set wrapper's version
fieldLastVersion.set(objectWrapper, objectVersion);
}
}
}
注册订阅：
LiveDataBus.get()
.with("key_test", String.class)
.observe(this, new Observer<String>() {
@Override
public void onChanged(@Nullable String s) {
}
});
发送消息：
LiveDataBus.get().with("key_test").setValue(s);
