Android 使用 LiveData 实现 EventBus

8,006 阅读14分钟

绪论

本文是学习了大佬的文章后,自己去动手实践后写的一篇学习笔记。大佬的文章写得比较好,我自己去写未必描述得那么清楚😂,所以本文很多地方都直接引用了大佬的文章。

项目源码:github.com/LinYaoTian/…

Tip:阅读本文最好对 Jetpack 的 LIfeCycle 和 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 的作用

  • 管理组件的生命周期。
  • 让第三方业务能在自己内部就能拿到依赖的组件的生命周期,便于及时叫停,避免错过执行时机。

想继续了解的同学可以看下这两篇文章:

下面继续说 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 是业界知名的通信类总线库,但它也存在许多被人诟病的缺点:

  1. 需要手动的注册和反注册,稍不小心可能会造成内存泄露。
  2. 使用 EventBus 出错时难以跟踪出错的事件源。
  3. 每个事件都要定义一个事件类,容易造成类膨胀。

而通过 LiveData 实现的 LiveDataBus 的具有以下优点:

  1. 具有生命周期感知能力,不用手动注册和反注册。
  2. 具有唯一的可信事件源。
  3. 以字符串区分每一个事件,避免类膨胀。
  4. 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,所以订阅者会马上收到订阅之前发布的最新消息!!

问题二

我们看一下 LiveData 的 postValue() 方法的注释:

可以看到,注释中说,如果在多线程中同一个时刻,多次调用了 postValue() 方法,只有最后次调用的值会得到更新。也就是此方法是有可能会丢失事件!!!

看下源码:

android.arch.lifecycle.LiveData

    private final Runnable mPostValueRunnable = new Runnable() {
        @Override
        public void run() {
            Object newValue;
            synchronized (mDataLock) {
                newValue = mPendingData;
              	// 设置 mPendingData 为 not_set
                mPendingData = NOT_SET;
            }
            setValue((T) newValue);
        }
    };

    protected void postValue(T value) {
        boolean postTask;
        synchronized (mDataLock) {
          	// 判断 postTask 是否为 not_set
            postTask = mPendingData == NOT_SET;
            mPendingData = value;
        }
        if (!postTask) {
            return;
        }
        ArchTaskExecutor.getInstance().postToMainThread(mPostValueRunnable);
    }

从上面的源码就可以很容易看出,postValue 只是把传进来的数据先存到 mPendingData,然后往主线程抛一个 Runnable,在这个 Runnable 里面再调用 setValue 来把存起来的值真正设置上去,并回调观察者们。而如果在这个 Runnable 执行前多次 postValue,其实只是改变暂存的值 mPendingData,并不会再次抛另一个 Runnable。

最终实现

明白了问题根源所在,我们就好解决问题了。

解决这个问题的方案有多种,其中美团大佬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]")
    }
  	
  	override fun postValue(value: T) {
       mMainHandler.post {
           setValue(value)
       }
    }

    @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 默认实现即可。

同时重写 postValue() 方法,内部通过 MainHandler.post() 的方式去执行 setValue() 去避免 postValue() 会丢失数据的问题。

第三步:定义管理类

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 的源码:github.com/LinYaoTian/…

如果文章有任何问题,欢迎在评论区指正。