JetPack | LiveData 如何安全的观察数据
LiveData 是什么？LiveData是JetPack组件之一，LiveData是一个可观察的数据持有类，可以感知生命周期。是一种可观察的数据存储器类。与常规的可观察类不同，LiveData 具有生命周期感知能力，意指它遵循其他应用组件（如 Activity、Fragment 或 Service）的生命周期。这种感知能力可确保 LiveData 仅更新处于活跃生命周期状态的应用组件观察者。（来自Android官方解释）
LiveData的介绍和使用不在累述，直接看官方文档，本篇文章旨在讲解LiveData存在的意义以及实现的原理。LiveData 为什么会出现？之前看过重学安卓的小专栏的讲解：
LiveData 它被设计为仅限于负责 数据在订阅者生命周期内的被分发，除了 setValue / postValue 发送数据，以及 observe订阅数据. 没有多余的方法。(摘自：重学Android KunMinX-LiveData诞生的设计原因)
LiveData 只有如下简单的几个类：
postValue/setValue
在LiveData只存在两个方法postValue/setValue来进行数据的分发，那么这两个方法有什么区别呢？
postValue: 可以在任意的线程下执行。setValue: 只能在主线程下执行。
//<T> 决定livedata 持有的数据类型
liveData = MutableLiveData<String>()
//设置持有的数据
//postValue 可以在任意的线程下执行
liveData.postValue("1")
thread {
liveData.postValue("3")
}
//setValue 只能在主线程执行
liveData.value = "2"LiveData中源码的实现如下：实现的方式非常简单，最终都会调用到setValue最终存储在mData
protected void postValue(T value) {
boolean postTask;
synchronized (mDataLock) {
postTask = mPendingData == NOT_SET;//NOT_SET 是一个空对象
mPendingData = value;//存储发送的数据
}
if (!postTask) {//如果mPendingData 不等于 NOT_SET 说明mPostValueRunnable还没有执行完毕
return;
}
//主线程执行mPostValueRunnable
ArchTaskExecutor.getInstance().postToMainThread(mPostValueRunnable);
}
//主线程执行Runnable
private final Runnable mPostValueRunnable = new Runnable() {
@SuppressWarnings("unchecked")
@Override
public void run() {
Object newValue;
synchronized (mDataLock) {
newValue = mPendingData;
mPendingData = NOT_SET;//将mPendingData置为空对象 对应了上述的postTask
}
setValue((T) newValue);//最终还是调用了setValue()方法
}
};
@MainThread //注解 标注了在主线程执行
protected void setValue(T value) {
assertMainThread("setValue"); // 判断是否在主线程
mVersion++;//版本号 后续会用到
mData = value;//mData 真实的数据
dispatchingValue(null);//将数据分发给观察者
}发布/订阅数据
从上述的postValue/setValue 的方法，我们可以看到最终都会调用到
setValue并且调用了dispatchingValue进行发布数据。
LiveData中订阅数据,通过observe方法实现订阅数据：
//注册订阅者
//LifecycleOwner AppCompatActivity进行了实现
liveData.observe(this, {
Log.e("liveData-1", "onCreate: $it")
})
我们先来看分发数据dispatchingValue是如何实现的：(先关注分发数据的代码)
void dispatchingValue(@Nullable ObserverWrapper initiator) {
if (mDispatchingValue) {
mDispatchInvalidated = true;
return;
}
mDispatchingValue = true;
do {
mDispatchInvalidated = false;
if (initiator != null) {//TODO 这里先不要关注 后面会讲解
considerNotify(initiator);
initiator = null;
} else {
for (Iterator<Map.Entry<Observer<? super T>, ObserverWrapper>> iterator =
mObservers.iteratorWithAdditions(); iterator.hasNext(); ) {//TODO 关注这里 遍历观察者
considerNotify(iterator.next().getValue());//TODO 通知观察者有数据过来了
if (mDispatchInvalidated) {
break;
}
}
}
} while (mDispatchInvalidated);
mDispatchingValue = false;
}
private void considerNotify(ObserverWrapper observer) {
//其他代码先不要关注
....
observer.mObserver.onChanged((T) mData);//将数据通知观察者 将mData传递过去
}
如下图所示：最终setValue通过considerNotify将数据通知给订阅者。
observe订阅数据：
@MainThread //看这里这里标注了 observe必须在主线程中调用
public void observe(@NonNull LifecycleOwner owner, @NonNull Observer<? super T> observer) {
assertMainThread("observe");//判断是否在主线程中
//感知生命周期 先忽略后面讲解
if (owner.getLifecycle().getCurrentState() == DESTROYED) {
// ignore
return;
}
//感知生命周期 先忽略后面讲解
LifecycleBoundObserver wrapper = new LifecycleBoundObserver(owner, observer);
//TODO 这里将observer存储在mObservers中。这里mObservers，通过dispatchingValue调用分发数据
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;
}
owner.getLifecycle().addObserver(wrapper);
}注意：observer 必须在主线程中调用。
那么整体下来，LiveData分发数据和订阅数据的流程图如下：
生命周期管理
在JetPack组件通过Lifecycle来管理生命周期，关于Lifecycle我会单独出一篇文章讲解。关于Lifecycle先看AppCompatActivity实现了LifecycleOwner.可以获取getLifecycle,通过getLifecycle.addObserver()来注册生命周期的观察者
MyLifeCycle实现了LifecycleEventObserver，生命周期的观察者
class MyLifeCycle : LifecycleEventObserver {
override fun onStateChanged(source: LifecycleOwner, event: Lifecycle.Event) {
Log.e("TAG", "onStateChanged: " + source.lifecycle.currentState+" event:"+event)
}
}
在Activity中注册观察者：
//感知Activity的生命周期 注册生命周期的观察者
lifecycle.addObserver(MyLifeCycle())
@MainThread
public void observe(@NonNull LifecycleOwner owner, @NonNull Observer<? super T> observer) {
assertMainThread("observe");//判断是否在主线程中
//获取当前生命周期的状态如果destory状态直接返回
if (owner.getLifecycle().getCurrentState() == DESTROYED) {
// ignore
return;
}
//LifecycleBoundObserver 实现了LifecycleEventObserver
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;
}
//注册生命周期的观察者 owner: Activity/Fragment
owner.getLifecycle().addObserver(wrapper);
}那么生命周期感知的关键就是LifecycleBoundObserver类的实现：
class LifecycleBoundObserver extends ObserverWrapper implements LifecycleEventObserver {
@NonNull
final LifecycleOwner mOwner;
//构造方法存储LifecycleOwner和Observer
LifecycleBoundObserver(@NonNull LifecycleOwner owner, Observer<? super T> observer) {
super(observer);
mOwner = owner;
}
//判断生命周期是否处于活跃状态
@Override
boolean shouldBeActive() {
//判断是否处于活跃状态 isAtLeast进行状态比较：compareTo(state) >= 0; 处于STARTED RESUMED
return mOwner.getLifecycle().getCurrentState().isAtLeast(STARTED);
}
// onStateChanged感知生命周期
@Override
public void onStateChanged(@NonNull LifecycleOwner source,
@NonNull Lifecycle.Event event) {
//获取当前生命周期的状态
Lifecycle.State currentState = mOwner.getLifecycle().getCurrentState();
//感知到生命周期状态destory移除
if (currentState == DESTROYED) {
removeObserver(mObserver);//从mObservers中移除mObserver 返回
return;
}
//感知到处于非destory状态
Lifecycle.State prevState = null;
while (prevState != currentState) {
prevState = currentState;
//判断是否处于生命周期活跃状态
activeStateChanged(shouldBeActive());
currentState = mOwner.getLifecycle().getCurrentState();
}
}
//判断是否为同一个LifecycleOwner
@Override
boolean isAttachedTo(LifecycleOwner owner) {
return mOwner == owner;
}
//移除该生命周期的观察者
@Override
void detachObserver() {
mOwner.getLifecycle().removeObserver(this);
}
}移除观察者：
@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);
}从上述代码中，可以看到，LifecycleBoundObserver类作为生命周期的观察者，主要有两个方法：
shouldBeActive判断是否处于活跃状态也就是处于：STARTED RESUMED状态。onStateChanged感知生命周期，如果感知到处于destory状态，则执行removeObserver(也标注了MainThread也就说必须在主线程中执行)移除数据的观察者以及生命周期的观察者。其他的状态通过activeStateChanged父类的方法处理。
LiveData 有两种观察者：一种是Observer数据的观察者，一种是LifecycleBoundObserver生命周期的观察者。
activeStateChanged的实现在ObserverWrapper父类中：
private abstract class ObserverWrapper {
final Observer<? super T> mObserver;//存储Observer
boolean mActive;//记录是否处于活跃状态 默认FALSE
int mLastVersion = START_VERSION;//最新版本 后续会用到
ObserverWrapper(Observer<? super T> observer) {
mObserver = observer;
}
//子类LifecycleBoundObserver
abstract boolean shouldBeActive();
//默认返回 FALSE 在子类中已经实现 LifecycleBoundObserver
boolean isAttachedTo(LifecycleOwner owner) {
return false;
}
//在子类中已经实现 LifecycleBoundObserver
void detachObserver() {
}
//活跃状态变更逻辑
void activeStateChanged(boolean newActive) {
if (newActive == mActive) {
return;
}
// immediately set active state, so we'd never dispatch anything to inactive
// owner
mActive = newActive;
changeActiveCounter(mActive ? 1 : -1);
if (mActive) {
//处于活跃状态 STARTED RESUMED 调用dispatchingValue分发数据
dispatchingValue(this);
}
}
}
在上述代码中，activeStateChanged最终判断mActive如果为TRUE 也就是活跃状态，则调用dispatchingValue并且传递了ObserverWrapper，分发数据。我们来验证一下生命周期感知,如下代码：5秒之后在发送数据，我们在5秒之内App进入后台，在进入前台，看是否能观察到数据。
理论的状态下是，App进入后台会进入STOPED状态，不会通知数据观察者，当App进入前台进入RESUMED状态，会通知观察者
//注册观察者
//LifecycleOwner AppCompatActivity进行了实现
liveData.observe(this, {
Log.e("liveData-1", "onCreate: $it")
})
liveData.observe(this, {
Log.e("liveData-2", "onCreate: $it")
})
//延时执行 app进入后台 10S在进入前台 查看数据
Handler(Looper.getMainLooper()).postDelayed({
liveData.postValue("lifecyle")
}, 5000)结果如下：注册了多个观察者都会在STARTED状态就收到了数据
注意：liveData.observe(...) 每调用一次observe，都会生成一个LifecycleBoundObserver对象，注册生命周期观察者，感知生命周期的变化。
之前看过dispatchingValue 在调用setValue和postValue传递参数都是null，而生命周期感知中传递了ObserverWrapper，来看看有什么不同。
void dispatchingValue(@Nullable ObserverWrapper initiator) {
if (mDispatchingValue) {
mDispatchInvalidated = true;
return;
}
mDispatchingValue = true;
do {
mDispatchInvalidated = false;
if (initiator != null) {
considerNotify(initiator);//通知数据观察者
initiator = null;
} else {
//..... setvalue/postvalue 逻辑上述讲过了
}
} while (mDispatchInvalidated);
mDispatchingValue = false;
}
private void considerNotify(ObserverWrapper observer) {
//如果没有处于活跃状态 则直接return
if (!observer.mActive) {
return;
}
//再一次判断是否处于活跃状态
if (!observer.shouldBeActive()) {
observer.activeStateChanged(false);
return;
}
//observer的版本和setValue的版本如果相等或者大于 则返回
if (observer.mLastVersion >= mVersion) {
return;
}
//统一版本号，版本号的作用在文章的后面
observer.mLastVersion = mVersion;
//通知观察者
observer.mObserver.onChanged((T) mData);
}整体的流程图如下：
粘性事件
通过上述的分析，了解了LiveData的具体的实现，目前还存在一个问题，在网上大部分文章都说LiveData支持粘性事件，那么什么是粘性事件呢?
思考：在下面代码中，先调用postValue在调用观察者，观察数据能监听到数据吗？
liveData.postValue("11111")
//粘性事件 liveData同理 注册生命周期的观察者
liveData.observe(this, {
Log.e("liveData-3", "onCreate: $it")
})答案是可以的，为什么呢？来看下面的一段代码，通过一个按钮来注册生命周期的观察者，来看会打印什么？
fun setViscous(view: View) {
//再次注册 生命周期的观察者 会打印当前的生命周期状态
lifecycle.addObserver(MyLifeCycle())
//粘性事件 liveData同理 注册生命周期的观察者
liveData.observe(this, {
Log.e("liveData-3", "onCreate: $it")
})
}看结果：原来如此，再次注册生命周期的观察者，会调用onStateChanged,那么在LiveData的原理是一样的啊，当注册数据观察者，同事也会注册生命周期的观察者，在LifecycleBoundObserver中感知生命周期的变化，调用了activeStateChanged，处于活跃状态，将最新的mData数据会返回给观察者。
但是粘性事件，存在着数据倒灌的现象：xiaozhuanlan.com/topic/67193…
LiveData中版本号的作用
在LiveData中持有一个mVersion版本号，在ObserverWrapper中持有一个mLastVersion的版本号，这两个版本号到底有什么作用呢？对于mVersion的变化是当调用setValue的时候才会+1
@MainThread
protected void setValue(T value) {
assertMainThread("setValue");
mVersion++;
mData = value;
dispatchingValue(null);
}对于mLastVersion 的变化,在调用considerNotify去通知数据观察者的时候才会发生改变。
private void considerNotify(ObserverWrapper observer) {
if (!observer.mActive) {
return;
}
if (!observer.shouldBeActive()) {
observer.activeStateChanged(false);
return;
}
//也就是说当mLastVersion==mVersion的时候不会在通知观察者
if (observer.mLastVersion >= mVersion) {
return;
}
observer.mLastVersion = mVersion;
observer.mObserver.onChanged((T) mData);
}也就是说当mLastVersion>=mVersion的时候不会在通知订阅者,可以这样理解，当setValue发生改变的时候mVersion > mLastVersion才会通知观察者。确保了只有setValue底层数据发生改变，LiveData才会通知订阅者。很有意思的设计，看源码果然可以学到很多优秀的思想
重复注册多个订阅者的问题
如下代码: 通过一个按钮，反复执行如下代码，根据LiveData的粘性事件，那么每次点击按钮都可以收到最新的消息
ld.observe(this){
Log.e("TAG", "setViscous: ${it}")
}
ld.observe(this,object :Observer<String>{
override fun onChanged(t: String?) {
Log.e("TAG", "onChanged: ${t}" )
}
})但是实际的结果，第一次两个都会收到，但是第二次只有下面的观察者才会收到。这是为什么呢？
这里的 “重复注册多个订阅者” 主要发生于 “使用匿名内部类而非 lambda” 的情况：当使用 lambda 时，基于 LiveData observe 方法内部的判断，在同一个页面或 adapter 内订阅不会发生重复订阅，但如果是匿名内部类，每次都会被认为是新的不同的实例，从而额外增加了一个新的订阅者。
LiveDataBus 以及防止数据倒灌
“数据倒灌”的概念来自-《重学安卓》@KunMinX 大佬提出的概念
LiveDataBus 实现LiveData跨组件通信，这里的跨组件指的是跨Activity和Fragment。但是LiveData由于粘性事件的设计会导致，addObserve 都会拿到最新的mData数据。
LiveDataBus的设计非常简单,通过单例模式存储LiveData. 数据倒灌的现象，通过图文和代码的方式来解释这一现象的问题：
1 . 这是在LiveDataActivity,在onCreate监听，上一个Activity的事件
val liveData = LiveDataBus.getInstance()?.with("test")
liveData?.observe(this){
Log.e("TAG", "onGoTo: $it" )
}//按钮：跳转到LiveData2Activity
startActivity(Intent(this, LiveData2Activity::class.java))
2 . 在LiveData2Activity,也要通过该"test"存储的LiveData和LiveData3Activity进行通信
class LiveData2Activity : AppCompatActivity() {
private var liveData:UnPeekLiveData<Any>? = null
override fun onCreate(savedInstanceState: Bundle?) {
super.onCreate(savedInstanceState)
setContentView(R.layout.activity_live_data2)
val tv = findViewById<TextView>(R.id.tv)
liveData = LiveDataBus.getInstance()?.with("test")
}
fun onValue(view: View) {
liveData?.postValue("test") //将数据发送给 LiveData3Activity
}
}3 . 但是当从LiveData2Activity返回，LiveDataActivity就出现了问题.在LiveDataActivity的观察者由于粘性事件，也就是当前的Activity处于了活跃状态，就会拿到最新的mData回调给观察者导致了如下结果：
LiveData2Activity本质上想和LiveData3Activity进行通信，但是返回LiveDataAcivity的时候由于粘性事件导致了LiveDataActivity收到了不可预期的结果,这个结果本来不是给LiveDataActivity的。
数据倒灌如何解决呢？
- 通过反射的方式，去修改mLastVersion (不推荐使用)
- 通过包装Observer的方式去解决 (推荐github.com/KunMinX/UnP…)
在上述提到了LiveData中版本号的设计和作用，那么在这里就可以用到了，其实在UnPeek-LiveData就是通过版本号对比的方式进行解决：
代码核心的逻辑如下：
/**
* LiveDataBus 用于定义全局的LiveData 实现跨组件通信
*/
class LiveDataBus {
private var warehouse: HashMap<String, UnPeekLiveData<Any>> = HashMap()
companion object {
private var liveDataBus: LiveDataBus? = null
fun getInstance(): LiveDataBus? {
if (liveDataBus == null) {
synchronized(LiveDataBus::class.java) {
if (liveDataBus == null) {
liveDataBus = LiveDataBus()
}
}
}
return liveDataBus
}
}
fun with(key: String): UnPeekLiveData<Any>? {
return if (!warehouse.containsKey(key)) {
val liveData = UnPeekLiveData<Any>()
warehouse[key] = liveData
liveData
} else {
warehouse.get(key = key)
}
}
}
/**
* 粘性事件的问题修复
* 重写observer onChange判断版本号-推荐的版本
*/
class UnPeekLiveData<T>:MutableLiveData<T>(){
private val currentVersion = AtomicInteger(START_VERSION)
companion object{
private const val START_VERSION = -1
}
/**
* 非粘性事件
*/
override fun observe(owner: LifecycleOwner, observer: Observer<in T>) {
super.observe(owner, ObserveWrapper(observer,currentVersion.get()))
}
/**
* 粘性事件
*/
fun observeSticky(owner: LifecycleOwner, observer: Observer<in T>){
super.observe(owner, ObserveWrapper(observer))
}
override fun setValue(value: T) {
currentVersion.getAndIncrement()
super.setValue(value)
}
inner class ObserveWrapper(private val mObserver: Observer<in T>, private val mVersion:Int = START_VERSION):Observer<T>{
override fun onChanged(t: T) {
if (currentVersion.get() > mVersion && t != null){
mObserver.onChanged(t)
}
}
override fun equals(o: Any?): Boolean {
if (this === o) {
return true
}
if (o == null || javaClass != o::class) {
return false
}
val that = o as UnPeekLiveData<*>.ObserveWrapper
return mObserver == that.mObserver
}
override fun hashCode(): Int {
return Objects.hash(mObserver)
}
}
}手写LiveData 掌握设计思想
深入了解了LiveData，最后手写一下LiveData加深理解。只需要写出核心的逻辑代码即可。
/**
* 手写LiveData实现
*/
class MyLiveData<T> {
private val START_VERSION = -1
private var mData:T?= null
private var mVersion:Int = START_VERSION
private var mPaddingData:T? = null
private val H = Handler(Looper.getMainLooper())
private var mObserves = ConcurrentHashMap<Observe<T>, ObserveWrapper>()
private val currentVersion = AtomicInteger(START_VERSION)
/**
* 发布数据 可以在任意线程
*/
fun postValue(t:T){
H.post {
setValue(t)
}
}
/**
* 发布数据 只能在主线程
*/
@MainThread
fun setValue(t:T){
//判断是否在主线程
if (Thread.currentThread() != Looper.getMainLooper().thread) {
//如果不在主线程 抛出异常
throw IllegalAccessException("setValue 必须要在主线程中调用")
}
currentVersion.getAndIncrement()
mData = t
mVersion++
dispatchValue(null)
}
/**
* 订阅(粘性)消息
*/
@MainThread
fun addObserve(owner: LifecycleOwner,observe: Observe<T>){
val lifecycleObserve = LifecycleObserveWrapper(owner, observe)
mObserves[observe] = lifecycleObserve
owner.lifecycle.addObserver(lifecycleObserve)
}
/**
* 订阅非粘性消息
*/
@MainThread
fun addNoStickinessObserve(owner: LifecycleOwner,observe: Observe<T>){
this.addObserve(owner,NoStickinessObserve(observe,currentVersion.get()))
}
/**
* 分发数据
*/
private fun dispatchValue(wrapper: ObserveWrapper?){
if (wrapper == null){
mObserves.entries.forEach {
notifyObserve(it.value)
}
}else{
notifyObserve(wrapper)
}
}
/**
* 通知订阅
*/
private fun notifyObserve(observe:ObserveWrapper){
if (!observe.mActive){
return
}
if (!observe.shouldActive()){
observe.activeStateChange(false)
return
}
if (observe.mLastVersion >= mVersion){
return
}
observe.mLastVersion = mVersion
observe.observe.onChange(mData)
}
/**
* 监听生命周期
*/
inner class LifecycleObserveWrapper(private val lifecycleOwner: LifecycleOwner,observe: Observe<T>):ObserveWrapper(observe = observe),LifecycleEventObserver{
/**
* 判断activity是否处于活跃状态
*/
override fun shouldActive():Boolean{
return lifecycleOwner.lifecycle.currentState.isAtLeast(Lifecycle.State.STARTED)
}
override fun onStateChanged(source: LifecycleOwner, event: Lifecycle.Event) {
if (source.lifecycle.currentState == Lifecycle.State.DESTROYED){
return
}
activeStateChange(shouldActive())
}
}
abstract inner class ObserveWrapper(val observe: Observe<T>){
var mLastVersion = START_VERSION
var mActive:Boolean = false
abstract fun shouldActive():Boolean
fun activeStateChange(mActive: Boolean){
if (this.mActive == mActive){
return
}
this.mActive = mActive
if (mActive){
dispatchValue(this)
}
}
}
/**
* 监听回调
*/
interface Observe<T>{
fun onChange(value:T?)
}
/**
* 没有粘性事件的监听回调
*/
inner class NoStickinessObserve<T>(private val observe: Observe<T>, private val version:Int = START_VERSION):Observe<T>{
override fun onChange(value: T?) {
if (currentVersion.get() > version){
observe.onChange(value)
}
}
}
}思考
在看LiveData的源码过程中，有这样一段代码：来判断生命周期是否处于活跃的状态，那么Lifecycle.State和Lifecycle.Event 对应的状态是如何呢？为什么下面这段代码可以判断处于活跃状态？
lifecycleOwner.lifecycle.currentState.isAtLeast(Lifecycle.State.STARTED)