美团是如何使用LiveData搭建通信框架的？

美团是如何使用LiveData搭建通信框架的？

前言

之前的文章中我们介绍了LiveData的内部实现，并提出LiveData也可用作应用程序内的通信手段。而目前，网络上也确实有很多相关的实现文章。在众多实现中，来自美团的实现是质量最有保证的。但是美团的实现不止是使用LiveData，还涉及广播、跨进程通信等；本文将从美团的实现中抽取LiveData相关的内容，重新构建一个通信框架（毕竟跨进程等通信手段并不是每个项目都会使用到）。如果有对LiveData内部机制不了解的同学，强烈建议先阅读上篇文章《LiveData为何这么香，这些你都知道吗？》再来看本文！

推荐

文章将率先在公众号「Code满满」与个人博客「李益的小站」上发布，如果本文对你有帮助，就关注一下公众号吧！

一、需要做什么

在开始正式的代码讲解前，我们需要知道我们要做什么。在LiveData中共有两种订阅方式：observe()与observeForever()。其中observeForever()的订阅非常简单，观察者永远接收消息，中间不做其他判断，基本没有什么可以改动的地方。而observe()会对观察者所在页面的状态进行判断，如果页面处于活跃状态，则接收数据更新，否则不接收。显而易见，observe()对页面状态进判断的部分，是我们可以放手施为的地方。

了解了LiveData的两种订阅方式后，我们再来思考一下我们的通信框架应该如何搭建：

• 为了方便使用，这个框架最好是个单例
• 为了可以灵活控制页面是否处于活跃状态（这会影响观察者在何时可以接收到消息），需要重写LiveData中对页面活跃状态的判断
• 为了灵活的控制事件，需要有个池子来存储事件，这个池子可以是Map，当然有存储也肯定需要有删除
• LiveData原来只是提供了observe与observeForever方法，缺少类似EventBus中的粘性事件，我们需要补上

以上大致就是我们需要在搭建框架时需要注意的一些点，下面我们一点一点来实现。

二、XLiveData

我们先从修改LiveData中对页面活跃状态的判断开始。我们新建一个类，命名为XLiveData，此类主要用于重写LiveData中判断页面是否处于活跃状态的相关方法，所以XLiveData需要继承于MutableLiveData。

public class XLiveData<T> extends MutableLiveData<T> {......
}

1. XLifecycleBoundObserver

在XLiveData中，定义一个继承于LifecycleBoundObserver的内部类，命名为XLifecycleBoundObserver，用于代替LiveData内部的LifecycleBoundObserver。

protected Lifecycle.State observerActiveLevel() {return STARTED;
}private class XLifecycleBoundObserver extends LifecycleBoundObserver {XLifecycleBoundObserver(@NonNull LifecycleOwner owner, Observer<T> observer) {super(owner, observer);}@Overrideboolean shouldBeActive() {Lifecycle().getCurrentState().isAtLeast(observerActiveLevel());}
}

XLifecycleBoundObserver与LifecycleBoundObserver相比，主要是重写了shouldBeActive()方法。LifecycleBoundObserver中的shouldBeActive()方法是当Activity处于onStart()与onPause()生命状态之间时，判定页面处于活跃状态，返回true。此处改为我们可以自行控制，更加灵活，例如我们可以改为Activity处于onCreate()与onStop()生命状态之间时，判定页面处于活跃状态，返回true。

@Override
boolean shouldBeActive() {Lifecycle().getCurrentState().isAtLeast(observerActiveLevel());
}

2. 重写observe

既然重写了一个XLifecycleBoundObserver，那么LiveData的observe()方法也必须得重写，主要是将其中的LifecycleBoundObserver替换为XLifecycleBoundObserver。

public void observe(@NonNull LifecycleOwner owner, @NonNull Observer<? super T> observer) {//  super.observe(owner, observer);if (Lifecycle().getCurrentState() == DESTROYED) {// ignorereturn;}try {LifecycleBoundObserver wrapper = new XLifecycleBoundObserver(owner, (Observer<T>) observer);LifecycleBoundObserver existing = (LifecycleBoundObserver) callMethodPutIfAbsent(observer, wrapper);if (existing != null && !existing.isAttachedTo(owner)) {throw new IllegalArgumentException("Cannot add the same observer"+ " with different lifecycles");}if (existing != null) {return;}Lifecycle().addObserver(wrapper);} catch (Exception e) {e.printStackTrace();}
}

三、BusObservable

BusObservable是一个接口，定义了被观察者需要实现的方法。因为真正编写被观察者代码时，代码如果过多，阅读体验不会太好，影响我们对被观察者的理解，此处先用接口定义结构，方便我们理解。

public interface BusObservable<T> {/*** 发送消息** @param value 发送的消息*/void post(final T value);/*** 延迟发送消息** @param value 发送的消息* @param delay 延迟的毫秒数*/void postDelay(final T value, final long delay);/*** 延迟发送，带生命周期* 如果延时发送消息的时候sender处于非激活状态，消息取消发送** @param owner 消息发送者* @param value 发送的消息* @param delay 延迟毫秒数*/void postDelay(@NonNull final LifecycleOwner owner, final T value, final long delay);/*** 发送消息* 强制接收到消息的顺序和发送顺序一致** @param value 发送的消息*/void postOrderly(final T value);/*** 注册一个Observer，带有生命周期感知，自动取消订阅* 在注册之前的发送消息，在注册时不会被接收** @param owner    LifecycleOwner* @param observer 观察者*/void observe(@NonNull final LifecycleOwner owner, @NonNull final Observer<T> observer);/*** 注册一个Observer，带有生命周期感知，自动取消订阅* 在注册之前发送的消息，在注册时会被接收（消息同步）** @param owner    LifecycleOwner* @param observer 观察者*/void observeSticky(@NonNull final LifecycleOwner owner, @NonNull final Observer<T> observer);/*** 注册一个Observer，需手动解除绑定* 在注册之前的发送消息，在注册时不会被接收** @param observer 观察者*/void observeForever(@NonNull final Observer<T> observer);/*** 注册一个Observer，需手动解除绑定* 在注册之前发送的消息，在注册时会被接收（消息同步）** @param observer 观察者*/void observeStickyForever(@NonNull final Observer<T> observer);/*** 通过Forever方式注册的，需要调用该方法取消订阅** @param observer 观察者*/void removeObserver(@NonNull final Observer<T> observer);/*** 订阅了此BusObservable的Observer的所在页面的生命周期状态是否一直处于活跃状态** @param alwaysBeActive {@code true}-Observer可以在Activity的onCreate到onStop之间的生命周期状态接收消息*                       {@code false}-Observer可以在Activity的onStart到onPause之间的生命周期状态接收消息*/BusObservable<T> alwaysBeActive(final boolean alwaysBeActive);/*** 当BusObservable的所有的Observer都被移除时，BusObservable对应的Event是否从事件总线中移除** @param autoClear*/BusObservable<T> autoClear(final boolean autoClear);
}

四、LiveBusCore

在进行上述一系列铺垫后，我们可以开始编写核心的逻辑代码了。首先我们先自定义一个类，命名为LiveBusCore，此类将是我们构建的通信框架的核心代码所在。

1. ObserverWrapper

我们在LiveBusCore中，定义一个内部类ObserverWrapper，此类实现了Observer接口，主要用于代替LiveData的observe()与observeForever()方法中传入的Observer对象。ObserverWrapper在调用数据更新方法onChanged(T t)时，在内部比Observer多加了一层是否接受更新的判断。

private class ObserverWrapper<T> implements Observer<T> {@NonNullprivate Observer<T> observer;// 是否拒收消息private boolean isRejectEvent = false;private ObserverWrapper(@NonNull Observer<T> observer) {this.observer = observer;}@Overridepublic void onChanged(T t) {if (isRejectEvent) {isRejectEvent = false;return;}Changed(t);}
}

2. LiveEvent

在LiveBusCore中自定义一个内部类，命名为LiveEvent，此类会实现上述提到的BusObservable接口，所以LiveEvent就是我们的被观察者，也是我们通信框架中的事件对象。作为一个事件，其需要持有LiveData、用于标识自己的key以及一个用于存储自己观察者的Map。

private class LiveEvent<T> implements BusObservable<T> {@NonNullprivate final Object key;private final InternalLiveData<T> liveData;private final Map<Observer, ObserverWrapper<T>> observerMap;private final Handler mainHandler;public LiveEvent(@NonNull Object key) {this.key = key;this.liveData = new InternalLiveData<>();this.observerMap = new HashMap<>();this.mainHandler = new MainLooper());}......
}

2.1 InternalLiveData

虽然之前我们已经定义了一个XLiveData，但是XLiveData只是个基类，我们需要按照我们的实际需求再定义一个符合我们需求的LiveData。因为默认的LivData的观察者是在页面处于onStart()与onPause()之间才能收到消息，我们在最大可能的避免内存泄漏的情况下，可以增加一个使观察者接收消息的区间变为onCreate()与onStop()之间的选择，方便我们做更多的操作。

private class InternalLiveData<T> extends XLiveData<T> {private boolean observerAlwaysBeActive = false;private boolean autoClear = true;@Overrideprotected Lifecycle.State observerActiveLevel() {return observerAlwaysBeActive ? Lifecycle.State.CREATED : Lifecycle.State.STARTED;}@Overridepublic void removeObserver(@NonNull Observer<? super T> observer) {veObserver(observer);if (autoClear && !liveData.hasObservers()) {Instance().ve(key);}}
}

InternalLiveData除了重写了observerActiveLevel()方法，还重写了removeObserver()方法，添加了是否会自动从事件池中移除事件的配置。

2.2 发送消息

我们在LiveEvent中定义一个发送消息方法，如下：

@MainThread
private void postInternal(T value) {liveData.setValue(value);
}

上述私有方法是直接调用了LiveData的setValue方法，只能在主线程中调用，我们还需考虑到在非主线程中调用的情况，最后形式如下：

@Overridepublic void post(final T value) {if (isMainThread()) {postInternal(value);} else {mainHandler.post(new PostValueTask(value));}
}private class PostValueTask implements Runnable {private Object newValue;public PostValueTask(@NonNull Object newValue) {wValue = newValue;}@Overridepublic void run() {postInternal((T) newValue);}
}

2.3 普通订阅

知道了如何发送消息，我们再看下如何订阅消息。在接收消息时，我们需要使用ObserverWrapper来代替Observer传递给LiveData，然后做下判断，ObserverWrapper只有在时，isRejectEvent才为true；其意义在于观察者不接收订阅之前的消息，只接收订阅后的消息。

另外，对于订阅的调用，我们同样做了在主线程和非主线程中调用的处理。

@MainThread
private void observeInternal(@NonNull LifecycleOwner owner, @NonNull Observer<T> observer) {ObserverWrapper<T> observerWrapper = new ObserverWrapper<>(observer);observerWrapper.isRejectEvent = Version() > XLiveData.START_VERSION;liveData.observe(owner, observerWrapper);
}@Override
public void observe(@NonNull final LifecycleOwner owner, @NonNull final Observer<T> observer) {if (isMainThread()) {observeInternal(owner, observer);} else {mainHandler.post(new Runnable() {@Overridepublic void run() {observeInternal(owner, observer);}});}
}

2.4 粘性事件

粘性事件是我们在开发中非常常用的一种事件，LiveData并没有给我们实现，一般都会用observeForever()方法来代替实现，但是每次订阅后，我们还必须手动取消订阅，如果忘记取消，可能会引发内存泄漏问题，这是非常不友好的。我们此处可以手动实现一下：

@MainThread
private void observeStickyInternal(@NonNull LifecycleOwner owner, @NonNull Observer<T> observer) {ObserverWrapper<T> observerWrapper = new ObserverWrapper<>(observer);liveData.observe(owner, observerWrapper);
}@Override
public void observeSticky(@NonNull final LifecycleOwner owner, @NonNull final Observer<T> observer) {if (isMainThread()) {observeStickyInternal(owner, observer);} else {mainHandler.post(new Runnable() {@Overridepublic void run() {observeStickyInternal(owner, observer);}});}
}

可以看到，observeStickyInterna()方法与observeInternal()方法基本相同，唯一的区别就是observeStickyInterna()中未对observerWrapper的isRejectEvent属性做处理，即观察者可以收到订阅之前的消息。

2.5 removeObserver

移除观察者就比较容易了，只需要在移除LiveData的观察者时，同时从LiveEvent的观察者集合中移除掉对应的观察者。

@MainThread
private void removeObserverInternal(@NonNull Observer<T> observer) {Observer<T> realObserver;if (ainsKey(observer)) {realObserver = ve(observer);} else {realObserver = observer;}veObserver(realObserver);
}@Override
public void removeObserver(@NonNull final Observer<T> observer) {if (isMainThread()) {removeObserverInternal(observer);} else {mainHandler.post(new Runnable() {@Overridepublic void run() {removeObserverInternal(observer);}});}
}

3. 单例

上述内容已经介绍了消息发送，普通订阅、粘性事件以及移除观察者，在我们定义的被观察者接口中还有延迟发送消息、永久订阅、永久粘性订阅等方法，这些方法与我们上述的实现都是大同小异，原理都是相同的，我们此处就不再赘述。最后我们看下单例的实现：

public class LiveBusCore {public static LiveBusCore getInstance() {return Holder.DEFAULT_BUS;}private static class Holder {private static final LiveBusCore DEFAULT_BUS = new LiveBusCore();}private Map<Object, LiveEvent> mBus;private LiveBusCore() {this.mBus = new HashMap<>();}public synchronized <T> BusObservable<T> with(Object key, Class<T> eventType) {if (!ainsKey(key)) {mBus.put(key, new LiveEvent<T>(key));}return (BusObservable<T>) (key);}public synchronized void removeEvent(@NonNull Object key) {if (!ainsKey(key)) {ve(key);}}......
}

为了方便使用，我们可以再封装一层：

public final class LiveBus {public static <T> BusObservable<T> get(@NonNull Object key, Class<T> type) {Instance().with(key, type);}public static <T> BusObservable<T> get(@NonNull Class<T> eventType) {Instance().Name(), eventType);}public static BusObservable<Object> get(@NonNull Object key) {Instance().with(key, Object.class);}public static void removeEvent(@NonNull Object key) {Instance().removeEvent(key);}
}

使用示例：

// 发送消息
(Constants.LK_TEST2_POST).post("收到来自Test2的消息")// 接收消息
(Constants.LK_TEST2_POST, String::class.java).observe(this, object : Observer<String> {override fun onChanged(t: String?) {alObserveData.value = "普通订阅：${t}"}})

五、小结

使用LiveData搭建通信框架的讲解就到此结束了，有兴趣的同学可以去我的Github上下载源码，项目是 Fly-Android()。如果在项目中没有跨进程通信等特殊需求的话，建议自己使用LiveData搭建通信框架，这样更加灵活，易修改。最后给出美团的框架：LiveEventBus。