共计 12153 个字符,预计需要花费 31 分钟才能阅读完成。
绪论
本文是学习了大佬的文章后,自己去动手实践后写的一篇学习笔记。大佬的文章写得比较好,我自己去写未必描述得那么清楚????,所以本文很多地方都直接引用了大佬的文章。
源码:https://github.com/LinYaoTian…
效果图:
Tip:阅读本文最好对 Jetpack 的 LIfeCycle 和 LiveData 有初步的了解。
引用:
- Android 消息总线的演进之路:用 LiveDataBus 替代 RxBus、EventBus
- 基于 LiveData 实现事件总线思路和方案
- Android Jetpack 架构组件(二)一文带你了解 Lifecycle(使用篇)
- Android Jetpack 架构组件(四)一文带你了解 LiveData(使用篇)
- Android Jetpack 架构组件(五)一文带你了解 LiveData(原理篇)
基本概念
对于 Android 系统来说,消息传递是最基本的组件,每一个 App 内的不同页面,不同的组件都在进行消息传递。
消息传递既可以用于 Android 四大组件之前的通信,也可以用于异步线程和主线程之前的通信。
对于 Android 开发者来说,经常使用的消息传递方式有很多种,从最早的 Handler、BroadcastReceiver、接口回调,到最近几年的流行的通信总线类框架 EventBus、RxBus。
EventBus
说到 Android 的通信总线类框架就不得不提到 EventBus。
EventBus 是一个 Android 事件发布 / 订阅框架,通过解耦发布者和订阅者简化 Android 事件传递。EventBus 可以代替 Android 传统的 Intent、Handler、Broadcast 或接口回调,在 Fragment、Activity、Service 线程之间传递数据,执行方法。
EventBus 最大的特点就是:简洁、解耦。
在没有 EventBus 之前我们通常用广播来实现监听,或者自定义接口函数回调,有的场景我们也可以直接用 Intent 携带简单数据,或者在线程之间通过 Handler 处理消息传递。但无论是广播还是 Handler 机制远远不能满足我们高效的开发。EventBus 简化了应用程序内各组件间、组件与后台线程间的通信。EventBus 一经推出,便受到广大开发者的推崇。
现在看来,EventBus 给 Android 开发者世界带来了一种新的框架和思想,就是消息的发布和订阅。 这种思想在其后很多框架中都得到了应用。
订阅 / 发布模式
订阅发布模式定义了一种“一对多”的依赖关系,让多个订阅者对象同时监听某一个主题对象。这个主题对象在自身状态变化时,会通知所有订阅者对象,使它们能够自动更新自己的状态。
订阅 / 发布模式和观察者模式两者非常相似,个人觉得订阅 / 发布模式是观察者模式的一种增强版。
两者的区别有:
- 观察者模式下,观察者和被观察者之前直接存在依赖关系。
- 而订阅 / 发布模式下,订阅者和发布者之前多了一个调度中心,避免了订阅者和发布者的依赖关系。
详情可以看这篇文章:观察者模式 vs 发布 - 订阅模式
LiveData
本文是通过 LiveData 去实现 EventBus,这里先介绍下 LiveData。
LiveData 如同它的名字一样,是一个可观察的数据持有者,和常规的 observable 不同,LiveData 是可以具有生命周期感知的,这意味着它能够在 Activity、Fragment、Service 中正确的处理生命周期。
实际上 LiveData 并不能单独起作用,它依赖于 Jectpack 的 LifeCycle 组件,因为 LifeCycle 已经在 Activity 的父类中被封装好了,在我们使用 LiveData 的过程中是基本是无感知的,所以这里就简单提一下 LifeCycle 好了。
LifeCycle 的作用
- 管理组件的生命周期。
- 让第三方业务能在自己内部就能拿到依赖的组件的生命周期,便于及时叫停,避免错过执行时机。
想继续了解的同学可以看下这两篇文章:
- Android Jetpack 架构组件(二)一文带你了解 Lifecycle(使用篇)
- Android Jetpack 架构组件(三)一文带你了解 Lifecycle(原理篇)
下面继续说 LiveData:
LiveData 的数据源一般是 ViewModel,也可以是其他可以更新的 LiveData 组件。
一般我们使用 LiveData 的 observe()
,当数据更新后,LiveData 会通知它的所有活跃的观察者。与 RxJava 不同的,LiveData 只会通知活跃的观察者,例如 Activity 位于 Destroyed 状态时是不活跃的,因此不会收到通知。
当然我们也可以使用 LiveData 的 observerForever() 方法进行订阅,区别是 observerForever() 不会受到 Activity 等组件的生命周期的影响,只要数据更新就会收到通知。
LiveData 的简单使用:
public class MainActivity extends AppCompatActivity {
private static final String TAG="MainActivity";
@Override
protected void onCreate(Bundle savedInstanceState) {super.onCreate(savedInstanceState);
setContentView(R.layout.activity_main);
// 创建 LiveData 对象
MutableLiveData<String> mutableLiveData = new MutableLiveData<>();
// 开始订阅
mutableLiveData.observe(this, new Observer<String>() {
@Override
public void onChanged(@Nullable final String s) {Log.d(TAG, "onChanged:"+s);
}
});
// 更新数据,最终会回调上面的 onChange() 方法
mutableLiveData.postValue("Android 进阶三部曲");
}
}
LiveData 的更多详情可以参考这篇文章 Android Jetpack 架构组件(四)一文带你了解 LiveData(使用篇)
通过 LiveData 实现的优点是什么?
EventBus 是业界知名的通信类总线库,但它也存在许多被人诟病的缺点:
- 需要手动的注册和反注册,稍不小心可能会造成内存泄露。
- 使用 EventBus 出错时难以跟踪出错的事件源。
- 每个事件都要定义一个事件类,容易造成类膨胀。
而通过 LiveData 实现的 LiveDataBus 的具有以下优点:
- 具有生命周期感知能力,不用手动注册和反注册。
- 具有唯一的可信事件源。
- 以字符串区分每一个事件,避免类膨胀。
- LiveData 为 Android 官方库,更加可靠。
原理
LiveDataBus 的组成
- 消息: 消息可以是任何的 Object,可以定义不同类型的消息,如 Boolean、String。也可以定义自定义类型的消息。
- 消息通道: LiveData 扮演了消息通道的角色,不同的消息通道用不同的名字区分,名字是 String 类型的,可以通过名字获取到一个 LiveData 消息通道。
- 消息总线: 消息总线通过单例实现,不同的消息通道存放在一个 HashMap 中。
- 订阅: 订阅者通过 getChannel() 获取消息通道,然后调用 observe() 订阅这个通道的消息。
- 发布: 发布者通过 getChannel() 获取消息通道,然后调用 setValue() 或者 postValue() 发布消息。
LiveDataBus 原理图
简单实现
通过 LiveData 我们可以非常简单的实现一个事件发布 / 订阅框架:
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> getChannel(String target, Class<T> type) {if (!bus.containsKey(target)) {bus.put(target, new MutableLiveData<>());
}
return (MutableLiveData<T>) bus.get(target);
}
public MutableLiveData<Object> getChannel(String target) {return getChannel(target, Object.class);
}
}
没错,上面短短的 27 行代码我们就实现了一个事件发布 / 订阅框架!
问题
LiveData 一时使用一时爽,爽完了之后我们发现这个简易的 LiveDataBus 存在一个问题,就是 订阅者会收到订阅之前发布的消息,类似于粘性消息。对于一个消息总线来说,粘性消息和非粘性消息都是必须支持的,下面我们来看一下如何解决这个问题。
我们先看一下为什么会出现这个问题:
LiveData 的订阅方法
android.arch.lifecycle.LiveData
@MainThread
public void observe(@NonNull LifecycleOwner owner, @NonNull Observer<? super T> observer) {assertMainThread("observe");
// 当前绑定的组件 (activity or fragment) 状态为 DESTROYED 的时候, 则会忽视当前的订阅请求
if (owner.getLifecycle().getCurrentState() == DESTROYED) {return;}
// 转为带生命周期感知的观察者包装类
LifecycleBoundObserver wrapper = new LifecycleBoundObserver(owner, observer);
ObserverWrapper existing = mObservers.putIfAbsent(observer, wrapper);
// 对应观察者只能与一个 owner 绑定
if (existing != null && !existing.isAttachedTo(owner)) {
throw new IllegalArgumentException("Cannot add the same observer"
+ "with different lifecycles");
}
if (existing != null) {return;}
// lifecycle 注册
owner.getLifecycle().addObserver(wrapper);
}
可以看到,LiveData 内部会将我们的传入参数包装成 wrapper,然后存在一个 Map 中,最后通过 LifeCycle 组件添加观察者。
LiveData 的更新数据方法
LiveData 更新数据方式有两个,一个是 setValue() 另一个是 postValue(),这两个方法的区别是,postValue() 在内部会抛到主线程去执行更新数据,因此适合在子线程中使用;而 setValue() 则是直接更新数据。
这里就只看下 setValue() 方法就好了
android.arch.lifecycle.LiveData
@MainThread
protected void setValue(T value) {assertMainThread("setValue");
// 发送版本 +1
mVersion++;
mData = value;
// 信息分发
dispatchingValue(null);
}
记住这里的 mVersion,它本问题关键,每次更新数据都会自增,默认值是 -1。然后我们跟进下 dispatchingValue() 方法:
android.arch.lifecycle.LiveData
void dispatchingValue(@Nullable ObserverWrapper initiator) {
// mDispatchingValue 的判断主要是为了解决并发调用 dispatchingValue 的情况
// 当对应数据的观察者在执行的过程中, 如有新的数据变更, 则不会再次通知到观察者
// 所以观察者内的执行不应进行耗时工作
if (mDispatchingValue) {
mDispatchInvalidated = true;
return;
}
mDispatchingValue = true;
do {
mDispatchInvalidated = false;
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;
}
可以看到,无论条件判断,最终都会执行 considerNotify() 方法,所以我们继续跟进:
android.arch.lifecycle.LiveData
private void considerNotify(ObserverWrapper observer) {if (!observer.mActive) {return;}
if (!observer.shouldBeActive()) {observer.activeStateChanged(false);
return;
}
// 判断 version
if (observer.mLastVersion >= mVersion) {return;}
observer.mLastVersion = mVersion;
//noinspection unchecked
observer.mObserver.onChanged((T) mData);
}
终于到了最关键的时候了!!如果 ObserverWrapper 的 mLastVersion 小于 LiveData 的 mVersion,那么就会执行的 onChange() 方法去通知观察者数据已更新。
而 ObserverWrapper.mLastVersion 的默认值是 -1,LiveData 只要更新过数据,mVersion 就肯定会大于 -1,所以订阅者会马上收到订阅之前发布的最新消息!!
最终实现
明白了问题根源所在,我们就好解决问题了。
解决这个问题的方案有多种,其中美团大佬 Android 消息总线的演进之路:用 LiveDataBus 替代 RxBus、EventBus 使用的反射的方式修改 LiveData 中的 mVersion 值去实现。还有另一个方案基于 LiveData 实现事件总线思路和方案,此方案是基于自定义观察者包装类,因为粘性消息最终会调用到 Observer#onChange() 方法,因此我们自定义 Observer 包装类,自己维护实际的订阅消息数,来判断是否需要触发真正的 onChange() 方法。本文使用的是第二种方案。
第一步:自定义 Observer 包装类
为了业务的拓展性,这里我们定义先一个 Base 基类:
internal open class BaseBusObserverWrapper<T>(private val mObserver: Observer<in T>, private val mBusLiveData: BusLiveData<T>) : Observer<T> {
private val mLastVersion = mBusLiveData.version
private val TAG = "BaseBusObserverWrapper"
override fun onChanged(t: T?) {Logger.d(TAG,"msg receiver =" + t.toString())
if (mLastVersion >= mBusLiveData.version){
// LiveData 的版本号没有更新过,说明并没有新数据,只是因为
// 当前 Observer 的版本号比 LiveData 低导致的调用 onChange()
return
}
try {mObserver.onChanged(t)
}catch (e:Exception){Logger.e(TAG,"error on Observer onChanged() =" + e.message)
}
}
open fun isAttachedTo(owner: LifecycleOwner) = false
}
这里我们保存了 LiveData 的 mVersion 值,每次执行 onChange() 时都先判断一些 LiveData 是否更新过数据,如果没有则不执行观察者的 Observer.onChange() 方法。
然后定义两个子类,BusLifecycleObserver 和 BusAlwaysActiveObserver,其中 BusLifecycleObserver 用于 LiveData#observer() 方法订阅的事件,而 BusAlwaysActiveObserver 用于 LiveData#observerForever() 方法订阅的事件。
internal class BusLifecycleObserver<T>(private val observer: Observer<in T>, private val owner: LifecycleOwner, private val liveData: BusLiveData<T>)
: BaseBusObserverWrapper<T>(observer,liveData),LifecycleObserver{@OnLifecycleEvent(Lifecycle.Event.ON_DESTROY)
fun onDestroy(){liveData.removeObserver(observer)
owner.lifecycle.removeObserver(this)
}
}
对于 BusLifecycleObserver,在生命周期组件处于 Destroyed 时,需要把观察者移除。
internal class BusAlwaysActiveObserver<T>(private val mObserver: Observer<in T>, private val mBusLiveData: BusLiveData<T>)
: BaseBusObserverWrapper<T>(mObserver, mBusLiveData)
对于 BusAlwaysActiveObserver,观察者不受组件生命周期的影响,因此不需要在组件 Destroyed 时移除。
第二步:自定义 LiveData 类
注意:因为 LiveData 的 getVersion() 是包访问级别的!所以 BusLiveData 必须定义到与 LiveData 同一个包内,即 androidx.lifecycle
包下,因此需要你自己创建一个同名的包并将 BusLiveData 放到里面。
如果你的项目工程没有引入 androidx,也可以使用 v7 包下的 android.arch.lifecycle
,方法同理,不过要注意的是 android.arch.lifecycle
包下 LiveData 的方法参数的泛型是没有型变的,因此直接复制这里代码会有点问题,需要你自己根据编译器的提示修改下。
class BusLiveData<T>(private val mKey:String) : MutableLiveData<T>() {
private val TAG = "BusLiveData"
private val mObserverMap: MutableMap<Observer<in T>, BaseBusObserverWrapper<T>> = mutableMapOf()
@MainThread
override fun observe(owner: LifecycleOwner, observer: Observer<in T>) {
val exist = mObserverMap.getOrPut(observer,{BusLifecycleObserver(observer,owner,this).apply {mObserverMap[observer] = this
owner.lifecycle.addObserver(this)
}
})
super.observe(owner, exist)
Logger.d(TAG,"observe() called with: owner = [$owner], observer = [$observer]")
}
@MainThread
override fun observeForever(observer: Observer<in T>) {super.observeForever(observer)
val exist = mObserverMap.getOrPut(observer ,{BusAlwaysActiveObserver(observer,this).apply {mObserverMap[observer] = this
}
})
super.observeForever(exist)
Logger.d(TAG, "observeForever() called with: observer = [$observer]")
}
@MainThread
fun observeSticky(owner: LifecycleOwner, observer: Observer<T>) {super.observe(owner, observer)
Logger.d(TAG, "observeSticky() called with: owner = [$owner], observer = [$observer]")
}
@MainThread
fun observeStickyForever(observer: Observer<T>){super.observeForever(observer)
Logger.d(TAG, "observeStickyForever() called with: observer = [$observer]")
}
@MainThread
override fun removeObserver(observer: Observer<in T>) {val exist = mObserverMap.remove(observer) ?: observer
super.removeObserver(exist)
Logger.d(TAG, "removeObserver() called with: observer = [$observer]")
}
@MainThread
override fun removeObservers(owner: LifecycleOwner) {mObserverMap.iterator().forEach {if (it.value.isAttachedTo(owner)) {mObserverMap.remove(it.key)
}
}
super.removeObservers(owner)
Logger.d(TAG, "removeObservers() called with: owner = [$owner]")
}
@MainThread
override fun onInactive() {super.onInactive()
if (!hasObservers()) {
// 当 LiveData 没有活跃的观察者时,可以移除相关的实例
LiveDataBusCore.getInstance().mBusMap.remove(mKey)
}
Logger.d(TAG, "onInactive() called")
}
@MainThread
public override fun getVersion(): Int {return super.getVersion()
}
}
代码比较短和简单,要点就是,对于非粘性消息,即 observer() 和 observerForever(),我们需要使用自定义的包装类包装处理。对于粘性消息,则直接使用 LiveData 默认实现即可。
第三步:定义管理类
internal class LiveDataBusCore {
companion object{
@JvmStatic
private val defaultBus = LiveDataBusCore()
@JvmStatic
fun getInstance() = defaultBus}
internal val mBusMap : MutableMap<String, BusLiveData<*>> by lazy {mutableMapOf<String, BusLiveData<*>>()
}
fun <T> getChannel(key: String) : BusLiveData<T> {return mBusMap.getOrPut(key){BusLiveData<T>(key)
} as BusLiveData<T>
}
}
第四步:定义入口类
class LiveDataBus {
companion object{
@JvmStatic
@Synchronized
fun <T> getSyn(key: String) : BusLiveData<T>{return get(key)
}
@JvmStatic
fun <T> get(key: String) : BusLiveData<T>{return LiveDataBusCore.getInstance().getChannel(key)
}
private fun <T> get(key: String, type: Class<T>) : BusLiveData<T> {return LiveDataBusCore.getInstance().getChannel(key)
}
@JvmStatic
fun <E> of(clz: Class<E>): E {require(clz.isInterface) {"API declarations must be interfaces."}
require(clz.interfaces.isEmpty()) {"API interfaces must not extend other interfaces."}
return Proxy.newProxyInstance(clz.classLoader, arrayOf(clz), InvocationHandler { _, method, _->
return@InvocationHandler get(
// 事件名以集合类名_事件方法名定义
// 以此保证事件的唯一性
"${clz.canonicalName}_${method.name}",
(method.genericReturnType as ParameterizedType).actualTypeArguments[0].javaClass)
}) as E
}
}
}
使用:
class TestLiveDataBusActivity : AppCompatActivity() {
companion object{private const val TAG = "TestLiveDataBusActivity"}
override fun onCreate(savedInstanceState: Bundle?) {super.onCreate(savedInstanceState)
setContentView(R.layout.activity_test_live_data_bus)
LiveDataBus.get<String>("testObserver").observe(this, Observer<String> {Log.d(TAG, "testObserver = $it")
test_observer_id.text = it
})
LiveDataBus.get<String>("testObserver").postValue("new value")
}
}
LiveDataBus 还提供了一个 of() 方法,用于提供用于事件约束,什么是时间约束呢?就是说我们的现在观察者和发布者获取消息渠道的 key 是一个字符串,在使用的过程中可能会出现,发布者的 key 是 itO 而观察者的 key 误输入成 it0 的情况,所以这里我们可以模仿 Retrofit 请求动态代理的做法,在使用的过程中,我们需要先定义一个接口:
interface TestLiveEvents {fun event1(): MutableLiveData<String>
}
使用:
fun main() {
LiveDataBus
.of(TestLiveEvents::class.java)
.event1()
.postValue("new value")
}
总结
借助于 Android 官方提供的 LiveData,我们可以非常方便地实现自己的 LiveDataBus,全部文件也就只有以上几个类,同时我们还避免了 EventBus 的许多缺点!
LiveDataBus 的源码:https://github.com/LinYaoTian…
如果文章有任何问题,欢迎在评论区指正。