Android JetPack LiveData源码解析

简介和使用

官方对LiveData的定义是：一种可观察的数据存储器类；具有以下优点：

• 采用观察者模式，无需在数据发生变化时更新界面便能确保界面符合数据状态；
• 具有生命周期感知能力，不需要手动处理生命周期，也不会因Activity停止而导致崩溃；
• 观察者绑定到了Lifecycle对象，会自动进行清理，不会发生内存泄露； 同一个观察者只能和一个Lifecycle绑定；
• 数据始终维持最新的状态，生命周期变为非活跃状态，它会在再次变为活跃状态时展示最新的数据（当Activity生命周期发生变化，重新回到活跃状态时会显示最新的数据）

基本使用方法：

创建：

class LiveDataViewModel : ViewModel() {
    val currentCount: MutableLiveData<Int> by lazy {
        MutableLiveData(0)
    }
}
复制代码

使用：

class LiveDataActivity : AppCompatActivity() {
    val viewModel: LiveDataViewModel by viewModels()
    override fun onCreate(savedInstanceState: Bundle?) {
        super.onCreate(savedInstanceState)
        setContentView(R.layout.activity_live_data)
        val textView = findViewById<TextView>(R.id.tv1)
        //添加监听，观察 LiveData 对象
        viewModel.currentCount.observe(this, Observer<Int> {
            textView.text = "$it"
        })
        findViewById<Button>(R.id.button).setOnClickListener {
        	//更新 LiveData 对象数据
            viewModel.currentCount.value = viewModel.currentCount.value?.plus(1)
        }

    }
}
复制代码

源码分析

开篇介绍的LiveData的优点：

• 观察者模式；
• 生命周期感知能力；
• 数据始终维持最新的状态。

我们主要从这三个方向对LiveData进行一下分析。

观察者

LiveData里对观察者的的定义很简单，接口只定义了一个接口：

public interface Observer<T> {
    void onChanged(T t);
}
复制代码

每当数据发生变化，onChanged 方法就会被调用。而Observer多以匿名内部类的方式实现，通过匿名内部类对onChanged方法的实现，完成数据变化改变之后的后置操作。

LiveData提供了两个方法用于注册观察者：

• observe(LifecycleOwner , Observer)
• observeForever(Observer)

两个方法的主要区别在于前者需要传如LifecycleOwner，用于提供了生命周期安全的保障，当组件被销毁后便会移除掉观察者。

所有注册的观察者都会保存在一个SafeIterableMap中，它是一个链表型结构，但是却提供Map键值对式的API。它在LiveData中声明并在声明时被赋值：

private SafeIterableMap<Observer<? super T>, ObserverWrapper> mObservers = 
	new SafeIterableMap<>();
复制代码

SafeIterableMap并不直接存储观察者，而是存储它们的包装类ObserverWrapper的实例。

observe(LifecycleOwner , Observer)和observeForever(Observer)都通过SafeIterableMap.putIfAbsent方法（以观察者Observer为Key，以包装类ObserverWrapper为value）添加数据：

public V putIfAbsent(@NonNull K key, @NonNull V v) {
    Entry<K, V> entry = get(key);
    if (entry != null) {
        return entry.mValue;
    }
    put(key, v);
    return null;
}
复制代码

该方法很简单，但有两个关键特点：

1. 如果所添加数据的key已经存在于链表中，则返回其对应的value；
2. 如果所添加数据的key不在于链表中，返回null。

也就是重复添加同一个Observer时，会直接返回第一次将它作为Key值添加时传入的ObserverWrapper实例。

添加(注册)观察者的方法有具体源码如下：

observe(LifecycleOwner , Observer)方法的代码如下：

public void observe(@NonNull LifecycleOwner owner, @NonNull Observer<? super T> observer) {
    //判断是否在主线程
    assertMainThread("observe");
    //当Lifecycle处于DESTROYED的状态时，不添加观察者，直接返回
    if (owner.getLifecycle().getCurrentState() == DESTROYED) {
        return;
    }
    //为观察者构建一个包装类
    LifecycleBoundObserver wrapper = new LifecycleBoundObserver(owner, observer);
    //将wrapper添加到链表中
    ObserverWrapper existing = mObservers.putIfAbsent(observer, wrapper);
    //如果观察者Observer已经被添加过了,并且该观察者
    //已经绑定到了其他LifecycleOwner，则直接抛出异常
    if (existing != null && !existing.isAttachedTo(owner)) {
        throw new IllegalArgumentException("Cannot add the same observer"
            + " with different lifecycles");
    }
    //观察者已经被添加并且已经绑定到了当前的LifecycleOwner
    if (existing != null) {
        return;
    }
    //添加Lifecycle观察者，实现对Lifecycle的监听
    owner.getLifecycle().addObserver(wrapper);
}
复制代码

该方法首先会对传入的观察者Observer进行一些校验操作，详细的操作已写在注释里了。由于该方法需要调用者传入LifecycleOwner实例，这意味着它具有感知生命周期的能力。事实它就是借助LifeCycle来判断观察者所依附的组件（Activity、Fragment）是否处于活跃状态，只有在活跃状态才会通知观察者数据变化。并且，当组件进入DESTROYED状态时，便会销毁观察者解除绑定以此避免可能会发生的内存泄露。

observeForever(Observer)方法相对比较简单：

public void observeForever(@NonNull Observer<? super T> observer) {
    //判断是否在主线程
    assertMainThread("observeForever");
    //为观察者构建一个包装类
    AlwaysActiveObserver wrapper = new AlwaysActiveObserver(observer);
    //将wrapper添加到链表中
    ObserverWrapper existing = mObservers.putIfAbsent(observer, wrapper);
    //如果观察者Observer已经被添加过了,并且该观察者
    //的包装类是LifecycleBoundObserver类型，也就是该观察者曾经
    //被作为参数调用过observe(LifecycleOwner , Observer)方法，则直接抛出异常
    if (existing instanceof LiveData.LifecycleBoundObserver) {
        throw new IllegalArgumentException("Cannot add the same observer"
            + " with different lifecycles");
    }
    if (existing != null) {
        return;
    }
    //主动触发activeStateChanged方法
    wrapper.activeStateChanged(true);
}
复制代码

详细流程也写在了注释里，就不再进行说明了。

总结如下：

• LiveData通过一个链表存储Observe观察者；
• LiveData提供了两个添加Observe的方法 ，一个生命周期安全和一个不安全的；
• Observe只能绑定一个生命周期组件，并且一旦绑定了之后就不能再调用observeForever方法。

生命周期感知能力

上一小节提到LifecycleBoundObserver和AlwaysActiveObserver的差别就在于是否和生命周期组件LifeCycle进行了绑定。而通过observe(LifecycleOwner , Observer)方法绑定的观察者具有生命周期感知能力。这里我们就通过源码看一下observe(LifecycleOwner , Observer)是如何为观察者赋予感知生命周期的能力的。

在上文分析observe(LifecycleOwner , Observer)和observeForever(Observer)添加观察者的代码我们知道，它们分别使用了LifecycleBoundObserver和AlwaysActiveObserver作为观察者的包装类，它们的差别就在于是否和生命周期组件LifeCycle进行了绑定。ObserverWrapper的源码如下：

private abstract class ObserverWrapper {

    //观察者实例
    final Observer<? super T> mObserver;
    //标记 mObserver 是否处于活跃状态
    boolean mActive;
    //标记数据版本号，用于和LiveData里的进行比对
    int mLastVersion = START_VERSION;

    //构造方法
    ObserverWrapper(Observer<? super T> observer) {
        mObserver = observer;
    }

    //判断观察者的宿主是否处于活跃状态（LifecycleOwner）
    abstract boolean shouldBeActive();

    //观察者是否已经绑定LifecycleOwner
    boolean isAttachedTo(LifecycleOwner owner) {
        return false;
    }

    //移除观察者
    void detachObserver() {
    }

    void activeStateChanged(boolean newActive) {
        if (newActive == mActive) {
            return;
        }
        mActive = newActive;
        changeActiveCounter(mActive ? 1 : -1);
        //如果持有的观察者mObserver处于活跃状态，则通过dispatchingValue
        //向观察者发送新数据
        if (mActive) {
            dispatchingValue(this);
        }
    }
}
复制代码

ObserverWrapper是一个抽象类，它持有观察者Observer的实例，并定义好了抽象方法和公用逻辑。这些方法主要用来判断观察者的状态。同时，作为观察者实现对LifeCycle的监听。进行生命周期的回调和处理。

在使用observe(LifecycleOwner , Observer)方法添加观察者时，使用的是ObserverWrapper的LifecycleBoundObserver子类：

class LifecycleBoundObserver extends ObserverWrapper implements LifecycleEventObserver {
    @NonNull
    final LifecycleOwner mOwner;

    LifecycleBoundObserver(@NonNull LifecycleOwner owner, Observer<? super T> observer) {
        super(observer);
        mOwner = owner;
    }

    @Override
    boolean shouldBeActive() {
        //当LifecycleOwner的LifeCycle处于STARTED和RESUMED时才认为
        //观察者处于活跃状态
        return mOwner.getLifecycle().getCurrentState().isAtLeast(STARTED);
    }

    @Override
    public void onStateChanged(@NonNull LifecycleOwner source,
        @NonNull Lifecycle.Event event) {
        //每当LifecycleOwner的生命周期发生变化时，该方法会被回调
        //当LifecycleOwner的LifeCycle处于DESTROYED状态时接触观察者
        Lifecycle.State currentState = mOwner.getLifecycle().getCurrentState();
        if (currentState == DESTROYED) {
            removeObserver(mObserver);
            return;
        }

        //不断比对当前状态，获取当前的生命周期状态，
        //并调用activeStateChanged
        Lifecycle.State prevState = null;
        while (prevState != currentState) {
            prevState = currentState;
            activeStateChanged(shouldBeActive());
            currentState = mOwner.getLifecycle().getCurrentState();
        }
    }

    @Override
    boolean isAttachedTo(LifecycleOwner owner) {
        //如果mOwner不为空，则说明观察者已经绑定了LifecycleOwner
        return mOwner == owner;
    }

    @Override
    void detachObserver() {
        //接触对生命周期组件的监听
        mOwner.getLifecycle().removeObserver(this);
    }
}
复制代码

LifecycleBoundObserver实现了LifecycleEventObserver接口，这使得它在observe(LifecycleOwner , Observer)方法中能通过owner.getLifecycle().addObserver(wrapper)实现对Lifecycle绑定。这里又是一个观察者模式，LifecycleBoundObserver实现了对生命周期组件Lifecycle的监听。每当Lifecycle发生改变，LifecycleBoundObserver.onStateChanged 方法就会被调用。当生命周期状态为DESTROYED时，则调用解除LiveData监听的removeObserver方法：

public void removeObserver(@NonNull final Observer<? super T> observer) {
    //判断是否在主线程
    assertMainThread("removeObserver");
    //调用SafeIterableMap的removed方法
    ObserverWrapper removed = mObservers.remove(observer);
    if (removed == null) {
        return;
    }
    //移除ObserverWrapper
    removed.detachObserver();
    removed.activeStateChanged(false);
}


//该方法通过遍历解除所有绑定
public void removeObservers(@NonNull final LifecycleOwner owner) {
    assertMainThread("removeObservers");
    for (Map.Entry<Observer<? super T>, ObserverWrapper> entry : mObservers) {
        if (entry.getValue().isAttachedTo(owner)) {
            removeObserver(entry.getKey());
        }
    }
}

复制代码

removeObserver在解除LiveData的观察者之后，会调用ObserverWrapper的detachObserver方法，该方法在LifecycleBoundObserver中的实现就是解除其对生命周期组件LifeCycle的监听。

observeForever(Observer)方法所使用的AlwaysActiveObserver就简单的多了：

private class AlwaysActiveObserver extends ObserverWrapper {

    AlwaysActiveObserver(Observer<? super T> observer) {
        super(observer);
    }

    @Override
    boolean shouldBeActive() {
        return true;
    }
}
复制代码

shouldBeActive固定返回true，就意味着只要数据变化就通知观察者。

总结如下：

LiveData通过一个包装类持有自己的观察者，并通过该包装类的实例实现对生命周期组件的监听，以此实现对生命周期感知型观察者的管理。不仅会在生命周期发生改变时通知数据变化（eg：设备旋转，会立即向观察者发送最新的可用数据），并且会在生命周期为DESTROYED状态时对贯彻着解除绑定，以此实现生命周期的感知能力。

维持数据的状态

LiveData不仅仅会在数据发生变化时通知已注册的观察者，在观察者注册时就有可能触发回调，将最新值返回给观察者：

例如下面的代码：

viewModel.currentCount.observe(this, Observer<Int> {
    textView1.text = "$it"
})
findViewById<Button>(R.id.button).setOnClickListener {
            //更新 LiveData 对象数据
    viewModel.currentCount.value = viewModel.currentCount.value?.plus(1)
    viewModel.currentCount.observe(this, Observer<Int> {
        textView2.text = "$it"
    })
}
复制代码

虽然textView2是在LiveData的数据改变后才注册了观察者进行了监听。但它依然可以拿到和textView1 同样的值。

首先看更新值时的逻辑：

数据更新

LiveData提供了两个方法用于更新值：

• setValue：只能从主线程/UI调用
• postValue：可以在非UI线程中调用

setValue的代码如下：

//标记LiveData正在给观察者发送数据
private boolean mDispatchingValue;

//标记发送中的数据已经失效
private boolean mDispatchInvalidated;

protected void setValue(T value) {
    assertMainThread("setValue");
    //更新Value的版本号用于和观察者里的进行比对
    mVersion++;
    mData = value;
    dispatchingValue(null);
}

void dispatchingValue(@Nullable ObserverWrapper initiator) {
    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;
}

private void considerNotify(ObserverWrapper observer) {
    //如果观察者处于非活跃状态，直接返回不处理
    if (!observer.mActive) {
        return;
    }
    if (!observer.shouldBeActive()) {
        observer.activeStateChanged(false);
        return;
    }

    //比对数据版本号，判断是否需要更新
    if (observer.mLastVersion >= mVersion) {
        return;
    }
    //同步数据版本号
    observer.mLastVersion = mVersion;
    //通知数据更新
    observer.mObserver.onChanged((T) mData);
}
复制代码

代码流程很简单，dispatchingValue中的判断数据是否处于发送中以及数据失效有必要讲一下。每当数据在更新的过程中，通过mDispatchingValue、mDispatchInvalidated分别实现对更新状态、数据是否失效的标记。从而实现对旧数据的舍弃和发送新的数据。典型的场景如下：

当有两个观察者A、B和C，在观察者A的onChanged方法里又调用了setValue方法。此时，如果使用setValue更新数据，LiveData被mDispatchingValue标记为发送数据状态，当A拿到数据后又会立即通过setValue触发dispatchingValue。这时，就会进入一轮新的循环向观察者发送新的数据。但是，这里就出现了一个问题，那就是只有观察者A收到了旧数据，B和C只收到了一次新数据。也就是说：LiveData会发生观察者收到数据的过程不同（可能有部分观察者漏掉一些数据），但是从结果看，它们最终都会拿到最新的数据。

用代码验证如下：

val currentCount: MutableLiveData<Int> by lazy {MutableLiveData()}

override fun onCreate(savedInstanceState: Bundle?) {
   super.onCreate(savedInstanceState)
   setContentView(R.layout.activity_main_messenger)
   
   currentCount.observe(this) {
   Log.e("A","------$it")
      if (it == 1){
         currentCount.value = 2
      }
   }
   currentCount.observe(this) {
      Log.e("B","------$it")
   }
   currentCount.observe(this) {
      Log.e("C","------$it")
   }
   currentCount.value = 1
}
复制代码

输出结果为：

A: ------1
A: ------2
B: ------2
C: ------2
复制代码

接着看postValue方法，它不强制调用者必须在主线程里。通过它可以在子线程里更新数据。

volatile Object mPendingData = NOT_SET;
private final Runnable mPostValueRunnable = new Runnable() {
    @SuppressWarnings("unchecked")
    @Override
    public void run() {
        Object newValue;
        synchronized (mDataLock) {
            newValue = mPendingData;
            mPendingData = NOT_SET;
        }
        setValue((T) newValue);
    }
};

protected void postValue(T value) {
    boolean postTask;
    //加锁确保数据安全
    synchronized (mDataLock) {
    	 //判断是否有数据更新
        postTask = mPendingData == NOT_SET;
        //确保暂存的值为最新
        mPendingData = value;
    }
    //如果有数据更新（说明数据处于更新中）则无需再次切换线程
    //直接通过更改mPendingData即可实现数据同步
    if (!postTask) {
        return;
    }
    ArchTaskExecutor.getInstance().postToMainThread(mPostValueRunnable);
}
复制代码

最终调用DefaultTaskExecutor的postToMainThread方法 ：

public void postToMainThread(Runnable runnable) {
    if (mMainHandler == null) {
        synchronized (mLock) {
            if (mMainHandler == null) {
                mMainHandler = createAsync(Looper.getMainLooper());
            }
        }
    }
      
    mMainHandler.post(runnable);
}
复制代码

代码相较于setValue就多了三点：

• 加锁确保线程安全；
• 通过Handler+ Runnable实现线程切换并调用setValue方法。接下来的流程就和setValue一样了。
• postValue 首先把传进来的数据存到mPendingData，然后再通过Runnable切换线程，在Runnable里面再调用setValue来把存起来的mPendingData发送给观察者们。由于mPendingData使用了volatile修饰，所以只需修改它的值无需启动Runable。这里就会造成数据丢失的情况。

如何确保最新的值

在上文中，不断的提到一个变量mVersion。它在LiveData中的声明和初始化如下：

public abstract class LiveData<T> {
    private int mVersion;
    public LiveData(T value) {
        mData = value;
        mVersion = START_VERSION + 1;
    }

    /**
     * Creates a LiveData with no value assigned to it.
     */
    public LiveData() {
        mData = NOT_SET;
        mVersion = START_VERSION;
    }
}
复制代码

它就是来标记数据的版本号的，每当一次调用一次setValue改变数据。它都会执行mVersion++操作：

protected void setValue(T value) {
    assertMainThread("setValue");
    mVersion++;
    mData = value;
    dispatchingValue(null);
}
复制代码

而在ObserverWrapper也维持着一个变量用来标记一个观察者拿到的数据的版本号：

private abstract class ObserverWrapper {
     int mLastVersion = START_VERSION;
}
复制代码

每当数据改变时，都会调用considerNotify方法来同步版本号：

private void considerNotify(ObserverWrapper observer) {
    if (!observer.mActive) {
        return;
    }
    if (!observer.shouldBeActive()) {
        observer.activeStateChanged(false);
        return;
    }
    if (observer.mLastVersion >= mVersion) {
        return;
    }
    observer.mLastVersion = mVersion;
    observer.mObserver.onChanged((T) mData);
}
复制代码

至于为什么在setValue调用之后注册的观察者也能拿到最新数据的原因observe(LifecycleOwner , Observer)方法和observeForever略有不同。

前者在添加生命周期监听owner.getLifecycle().addObserver(wrapper);时会触发LifecycleBoundObserver.onStateChanged进而触发activeStateChanged方法。而observeForever则更粗暴直接，它直接触发了activeStateChanged 方法：

public void observeForever(@NonNull Observer<? super T> observer) {
    assertMainThread("observeForever");
    AlwaysActiveObserver wrapper = new AlwaysActiveObserver(observer);
    ObserverWrapper existing = mObservers.putIfAbsent(observer, wrapper);
    if (existing instanceof LiveData.LifecycleBoundObserver) {
        throw new IllegalArgumentException("Cannot add the same observer"
            + " with different lifecycles");
    }
    if (existing != null) {
        return;
    }
    wrapper.activeStateChanged(true);
}
复制代码

小结

• LiveData通过setValue和postValue更新数据，前者不支持多线程；
• LiveData通过int值作为版本号和观察者比对，确保数据处于最新状态；
• LiveData只确保所有的观察者都能拿到最新的数据。setValue会存在部分观察者收不到过期数据，postValue方法会遗弃旧数据。

总结和思考

网上有很多关于ViewModel+LiveData替换RxJava的文章。这里说一下自己对LiveData和RxJava（RxAndroid）的理解。首先抛出自己的观点：它们两个职责的用处不同，严格的说不应该讨论谁替换谁的问题（至于为什么要加严格二字，我们文末再详细展开讨论）。

个人认为，Rx是为了让我们处理异步操作能像处理同步操作一样简单。

而Google官方则给了LiveData明确定义：

LiveData is an observable data holder class

LiveData就是一个可观察的数据存储器类！！而在上面我们也能发现，它之确保观察者拿到最新的数据。但不保证数据百分比传输，存在丢失数据和不同观察者拿到不同的回调的情况。

它们两个各有各的用处，不应该放在一起强行对比。Rx主要用来帮我们梳理逻辑、简化业务代码。LiveData则帮我们存储数据。

首先说一下RxJava，它全名是：Reactive Extensions for the JVM。可以理解为时Java的响应式编程扩展。它 不同于一般的库比如Retrofit、OkHttp等。它不为我们提供解决具体问题的实例，Rx给我们的是把逻辑变简单，让异步能像同步一样简单。而它的观察者Observable，你也可以将它理解为就是一个CallBack。不同之处在于上游数据生产者只管通过Observable发送数据，或是异步或是同步或者好几次切换。 下游观察者只管处理，也不应关心数据是怎么产生的。而且所有的回调，都在同一条线上，有点像一条河，或者流。

这时再回头看一下我们在Android中对它的使用：大多数是用来做个网络请求，切换一下线程。甚至连操作符都用不上几个。如果是这样，那么它当然可以被替换。但这是它的错吗？并不是，看一下Rx的设计初衷，你又是为什么使用它呢？你所看中的它的优点，是不是可以使用合理的代码设计去替换它呢？扪心自问，使用这个库之前有没有思考过其他方案？你有多大必要去使用它，又有多少跟风心理呢？如果你之是用来做一个网络请求，那么，它当然可以被替换。因为协程加LiveData可以做的更好，上手也更简单。

所以说在做技术选型时，一定要贴合业务和实际，不要盲目的引用。