9. 遥遥领先 Android jetpack全家桶三剑客之LiveData

随笔2周前发布勿念

6 0 0

1. LiveData干什么用的

2. LiveData的使用

3. LiveData的源码分析

LiveData 对比 Flow
LiveData使用会有的一些数据问题

6. 手写简单版本LiveData和liveDatabus

9. 遥遥领先 Android jetpack全家桶三剑客之LiveData

1. LiveData干什么用的

有感知能力的数据存储器

LiveData使得数据的更新能以观察者模式被observer感知

2个特点:

1) 、自动取消订阅： 当宿主生命周期进入消亡（DESTROYED）状态时，LiveData 会自动移除观察者，避免内存泄漏；
2) 、安全地回调数据： 在宿主生命周期状态低于活跃状态（STAETED）时，LiveData 不会回调数据，避免产生空指针异常或不必要的性能损耗；当宿主生命周期不低于活跃状态（STAETED）时，LiveData 会重新尝试回调数据，确保观察者接收到最新的数据。**

2. LiveData的使用

2.1 基本使用

创建LiveData实例，指定源数据类型
创建Observer实例，实现onChanged()方法，用于接收源数据变化并刷新UI
LiveData实例使用observe()方法添加观察者，并传入LifecycleOwner
LiveData实例使用setValue()/postValue()更新源数据（子线程要postValue()）

public class MutableLiveData<T> extends LiveData<T> {

    // 异步设置数据
    @Override
    public void postValue(T value) {
        super.postValue(value);
    }

    // 同步设置数据
    @Override
    public void setValue(T value) {
        super.setValue(value);
    }
}
public class LiveDataTestActivity extends AppCompatActivity{

   private MutableLiveData<String> mLiveData;
   
   @Override
   protected void onCreate(Bundle savedInstanceState) {
       super.onCreate(savedInstanceState);
       setContentView(R.layout.activity_lifecycle_test);
       
       //liveData基本使用
       mLiveData = new MutableLiveData<>();
       mLiveData.observe(this, new Observer<String>() {
           @Override
           public void onChanged(String s) {
               Log.i(TAG, "onChanged: "+s);
           }
       });
       Log.i(TAG, "onCreate: ");
       mLiveData.setValue("onCreate");//activity是非活跃状态，不会回调onChanged。变为活跃时，value被onStart中的value覆盖
   }
   @Override
   protected void onStart() {
       super.onStart();
       Log.i(TAG, "onStart: ");
       mLiveData.setValue("onStart");//活跃状态，会回调onChanged。并且value会覆盖onCreate、onStop中设置的value
   }
   @Override
   protected void onResume() {
       super.onResume();
       Log.i(TAG, "onResume: ");
       mLiveData.setValue("onResume");//活跃状态，回调onChanged
   }
   @Override
   protected void onPause() {
       super.onPause();
       Log.i(TAG, "onPause: ");
       mLiveData.setValue("onPause");//活跃状态，回调onChanged
   }
   @Override
   protected void onStop() {
       super.onStop();
       Log.i(TAG, "onStop: ");
       mLiveData.setValue("onStop");//非活跃状态，不会回调onChanged。后面变为活跃时，value被onStart中的value覆盖
   }
   @Override
   protected void onDestroy() {
       super.onDestroy();
       Log.i(TAG, "onDestroy: ");
       mLiveData.setValue("onDestroy");//非活跃状态，且此时Observer已被移除，不会回调onChanged
   }
}

感知只发生在 LifecycleOwner的活跃生命周期状态。

MutableLiveData

LiveData 通常会搭配 ViewModel 使用

2.2 高级使用

如果希望在将 LiveData 对象分派给观察者之前对存储在其中的值进行更改，或者需要根据另一个实例的值返回不同的 LiveData 实例，可以使用LiveData中提供的Transformations类。

2.2.1 数据修改 – Transformations.map

        //Integer类型的liveData1
        MutableLiveData<Integer> liveData1 = new MutableLiveData<>();
        //转换成String类型的liveDataMap
        LiveData<String> liveDataMap = Transformations.map(liveData1, new Function<Integer, String>() {
            @Override
            public String apply(Integer input) {
                String s = input + " + Transformations.map";
                Log.i(TAG, "apply: " + s);
                return s;
            }
        });
        liveDataMap.observe(this, new Observer<String>() {
            @Override
            public void onChanged(String s) {
                Log.i(TAG, "onChanged1: "+s);
            }
        });

        liveData1.setValue(100);

使用很简单：原本的liveData1 没有添加观察者，而是使用Transformations.map()方法对liveData1的数据进行的修改生成了新的liveDataMap，liveDataMap添加观察者，最后liveData1设置数据

2.2.2 数据切换 – Transformations.switchMap

如果想要根据某个值切换观察不同LiveData数据，则可以使用Transformations.switchMap()方法。

    //两个liveData，由liveDataSwitch决定 返回哪个livaData数据
        MutableLiveData<String> liveData3 = new MutableLiveData<>();
        MutableLiveData<String> liveData4 = new MutableLiveData<>();
        
    //切换条件LiveData，liveDataSwitch的value 是切换条件
        MutableLiveData<Boolean> liveDataSwitch = new MutableLiveData<>();
        
    //liveDataSwitchMap由switchMap()方法生成，用于添加观察者
        LiveData<String> liveDataSwitchMap = Transformations.switchMap(liveDataSwitch, new Function<Boolean, LiveData<String>>() {
            @Override
            public LiveData<String> apply(Boolean input) {
            //这里是具体切换逻辑：根据liveDataSwitch的value返回哪个liveData
                if (input) {
                    return liveData3;
                }
                return liveData4;
            }
        });

        liveDataSwitchMap.observe(this, new Observer<String>() {
            @Override
            public void onChanged(String s) {
                Log.i(TAG, "onChanged2: " + s);
            }
        });

        boolean switchValue = true;
        liveDataSwitch.setValue(switchValue);//设置切换条件值

        liveData3.setValue("liveData3");
        liveData4.setValue("liveData4");

2.2.3 观察多个数据 – MediatorLiveData

MediatorLiveData 是 LiveData 的子类，允许合并多个 LiveData 源。只要任何原始的 LiveData 源对象发生更改，就会触发 MediatorLiveData 对象的观察者

        MediatorLiveData<String> mediatorLiveData = new MediatorLiveData<>();

        MutableLiveData<String> liveData5 = new MutableLiveData<>();
        MutableLiveData<String> liveData6 = new MutableLiveData<>();

    //添加 源 LiveData
        mediatorLiveData.addSource(liveData5, new Observer<String>() {
            @Override
            public void onChanged(String s) {
                Log.i(TAG, "onChanged3: " + s);
                mediatorLiveData.setValue(s);
            }
        });
    //添加 源 LiveData
        mediatorLiveData.addSource(liveData6, new Observer<String>() {
            @Override
            public void onChanged(String s) {
                Log.i(TAG, "onChanged4: " + s);
                mediatorLiveData.setValue(s);
            }
        });

    //添加观察
        mediatorLiveData.observe(this, new Observer<String>() {
            @Override
            public void onChanged(String s) {
                Log.i(TAG, "onChanged5: "+s);
                //无论liveData5、liveData6更新，都可以接收到
            }
        });
        
        liveData5.setValue("liveData5");
        //liveData6.setValue("liveData6");

3. LiveData的源码分析

3.1 添加观察者

LiveData原理是观察者模式，下面就先从LiveData.observe()方法看起

    /**
     * 添加观察者. 事件在主线程分发. 如果LiveData已经有数据，将直接分发给observer。
     * 观察者只在LifecycleOwner活跃时接受事件，如果变为DESTROYED状态，observer自动移除。
     * 当数据在非活跃时更新，observer不会接收到。变为活跃时 将自动接收前面最新的数据。 
     * LifecycleOwner非DESTROYED状态时，LiveData持有observer和 owner的强引用，DESTROYED状态时自动移除引用。
     * @param owner    控制observer的LifecycleOwner
     * @param observer 接收事件的observer
     */
    @MainThread
    public void observe(@NonNull LifecycleOwner owner, @NonNull Observer<? super T> observer) {
        assertMainThread("observe");
        if (owner.getLifecycle().getCurrentState() == DESTROYED) {
            // LifecycleOwner是DESTROYED状态，直接忽略
            return;
        }
        //使用LifecycleOwner、observer 组装成LifecycleBoundObserver，添加到mObservers中
        LifecycleBoundObserver wrapper = new LifecycleBoundObserver(owner, observer);
        ObserverWrapper existing = mObservers中.putIfAbsent(observer, wrapper);
        if (existing != null && !existing.isAttachedTo(owner)) {
        //!existing.isAttachedTo(owner)说明已经添加到mObservers中的observer指定的owner不是传进来的owner，就会报错“不能添加同一个observer却不同LifecycleOwner”
            throw new IllegalArgumentException("Cannot add the same observer"
                    + " with different lifecycles");
        }
        if (existing != null) {
            return;//这里说明已经添加到mObservers中,且owner就是传进来的owner
        }
        owner.getLifecycle().addObserver(wrapper);
    }

3.2 事件回调

LiveData添加了观察者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() { //至少是STARTED状态
            return mOwner.getLifecycle().getCurrentState().isAtLeast(STARTED);
        }

        @Override
        public void onStateChanged(@NonNull LifecycleOwner source,
                @NonNull Lifecycle.Event event) {
            if (mOwner.getLifecycle().getCurrentState() == DESTROYED) {
                removeObserver(mObserver);//LifecycleOwner变成DESTROYED状态，则移除观察者
                return;
            }
            activeStateChanged(shouldBeActive());
        }

        @Override
        boolean isAttachedTo(LifecycleOwner owner) {
            return mOwner == owner;
        }

        @Override
        void detachObserver() {
            mOwner.getLifecycle().removeObserver(this);
        }
    }

最后观察者变为活跃，就使用LiveData的dispatchingValue(observerWrapper)进行数据分发:

    void dispatchingValue(@Nullable ObserverWrapper initiator) {
        if (mDispatchingValue) {
            mDispatchInvalidated = true;//如果当前正在分发，则分发无效，return
            return;
        }
        mDispatchingValue = true; //标记正在分发
        do {
            mDispatchInvalidated = false; 
            if (initiator != null) {
                considerNotify(initiator); //observerWrapper不为空，使用considerNotify()通知真正的观察者
                initiator = null;
            } else { //observerWrapper为空，遍历通知所有的观察者
                for (Iterator<Map.Entry<Observer<? super T>, ObserverWrapper>> iterator =
                        mObservers.iteratorWithAdditions(); iterator.hasNext(); ) {
                    considerNotify(iterator.next().getValue());
                    if (mDispatchInvalidated) {
                        break;
                    }
                }
            }
        } while (mDispatchInvalidated);
        mDispatchingValue = false; 
    }

分发的核心逻辑:

    void dispatchingValue(@Nullable ObserverWrapper initiator) {
        if (mDispatchingValue) {
            mDispatchInvalidated = true;//如果当前正在分发，则分发无效，return
            return;
        }
        mDispatchingValue = true; //标记正在分发
        do {
            mDispatchInvalidated = false; 
            if (initiator != null) {
                considerNotify(initiator); //observerWrapper不为空，使用considerNotify()通知真正的观察者
                initiator = null;
            } else { //observerWrapper为空，遍历通知所有的观察者
                for (Iterator<Map.Entry<Observer<? super T>, ObserverWrapper>> iterator =
                        mObservers.iteratorWithAdditions(); iterator.hasNext(); ) {
                    considerNotify(iterator.next().getValue());
                    if (mDispatchInvalidated) {
                        break;
                    }
                }
            }
        } while (mDispatchInvalidated);
        mDispatchingValue = false; 
    }

3.3 数据更新

LivaData数据更新可以使用setValue(value)、postValue(value)，区别在于postValue(value)用于子线程:

//LivaData.java
    private final Runnable mPostValueRunnable = new Runnable() {
        @SuppressWarnings("unchecked")
        @Override
        public void run() {
            Object newValue;
            synchronized (mDataLock) {
                newValue = mPendingData;
                mPendingData = NOT_SET;
            }
            setValue((T) newValue); //也是走到setValue方法
        }
    };

    protected void postValue(T value) {
        boolean postTask;
        synchronized (mDataLock) {
            postTask = mPendingData == NOT_SET;
            mPendingData = value;
        }
        if (!postTask) {
            return;
        }
        ArchTaskExecutor.getInstance().postToMainThread(mPostValueRunnable);//抛到主线程
    }

注册过程:

注册过程: 
      LifecycleOwner--->addObserver()
          LifecycleBoundObserver----->activeStateChanged () 
调用过程:  
                  livedata------->setValue
                          livedata------->dispatchingValue()
                                   livedata------->considerNotify()
                                            LifecycleBoundObserver----->activeStateChanged () // 进行分发

总结:

LivaData通过observe()添加与LifecycleOwner绑定的观察者；观察者变为活跃时回调最新的数据；使用setValue()、postValue()更新数据时会通知回调所有的观察者。

流程图:

9. 遥遥领先 Android jetpack全家桶三剑客之LiveData

5. LiveData使用会有的一些数据问题

1、LiveData 只能在主线程更新数据： 只能在主线程 setValue，即使 postValue 内部也是切换到主线程执行；
2、LiveData 数据重放问题： 注册新的订阅者，会重新收到 LiveData 存储的数据，这在有些情况下不符合预期（具体见第 TODO 节）；
3、LiveData 不防抖问题： 重复 setValue 相同的值，订阅者会收到多次 onChanged() 回调（可以使用 distinctUntilChanged() 优化）；
4、LiveData 丢失数据问题： 在数据生产速度 > 数据消费速度时，LiveData 无法观察者能够接收到全部数据。比如在子线程大量 postValue 数据但主线程消费跟不上时，中间就会有一部分数据被忽略。

5.1 LiveData 数据重放问题原理分析: (****重复收到数据)

LiveData 的数据重放问题也叫作数据倒灌、粘性事件，核心源码在 LiveData#considerNotify(Observer) 中：

首先，LiveData 和观察者各自会持有一个版本号 version，每次 LiveData#setValue 或 postValue 后，LiveData 持有的版本号会自增 1。在 LiveData#considerNotify(Observer) 尝试分发数据时，会判断观察者持有版本号是否小于 LiveData 的版本号（Observer#mLastVersion >= LiveData#mVersion 是否成立），如果成立则说明这个观察者还没有消费最新的数据版本。
而观察者的持有的初始版本号是 -1，因此当注册新观察者并且正好宿主的生命周期是大于等于可见状态（STARTED）时，就会尝试分发数据，这就是数据重放。

为什么 Google 要把 LiveData 设计为粘性呢？LiveData 重放问题需要区分场景来看 —— 状态适合重放，而事件不适合重放：

当 LiveData 作为一个状态使用时，在注册新观察者时重放已有状态是合理的；
当 LiveData 作为一个事件使用时，在注册新观察者时重放已经分发过的事件就是不合理的。

5.2) LiveData 数据重放问题解决方案:

方案一: **反射修改观察者版本号

业界分享出来的一个方案，不确定思路原创源。实现方法是在注册新观察者时，通过反射的手段将观察者持有的版本号（Observer#mLastVersion）同步为 LiveData 的版本号。缺点是使用反射，但确实能够解决多观察者问题。

private void hook(@NonNull Observer<T> observer) throws Exception {
    //get wrapper's version
    Class<LiveData> classLiveData = LiveData.class;
    Field fieldObservers = classLiveData.getDeclaredField("mObservers");
    fieldObservers.setAccessible(true);
    Object objectObservers = fieldObservers.get(this);
    Class<?> classObservers = objectObservers.getClass();
    Method methodGet = classObservers.getDeclaredMethod("get", Object.class);
    methodGet.setAccessible(true);
    Object objectWrapperEntry = methodGet.invoke(objectObservers, observer);
    Object objectWrapper = null;
    if (objectWrapperEntry instanceof Map.Entry) {
        objectWrapper = ((Map.Entry) objectWrapperEntry).getValue();
    }
    if (objectWrapper == null) {
        throw new NullPointerException("Wrapper can not be bull!");
    }
    Class<?> classObserverWrapper = objectWrapper.getClass().getSuperclass();
    Field fieldLastVersion = classObserverWrapper.getDeclaredField("mLastVersion");
    fieldLastVersion.setAccessible(true);
    //get livedata's version
    Field fieldVersion = classLiveData.getDeclaredField("mVersion");
    fieldVersion.setAccessible(true);
    Object objectVersion = fieldVersion.get(this);
    //set wrapper's version
    fieldLastVersion.set(objectWrapper, objectVersion);
}

方案二: UnPeekLiveData 反射方案优化

UnPeekLiveData 是 KunMinX 提出并开源的方案，主要思路是将 LiveData 源码中的 Observer#mLastVersion 和 LiveData#mVersion 在子类中重新实现一遍。在 UnPeekLiveData 中会有一个原子整型来标记数据版本，并且每个 Observer 在注册时会拿到当前 LiveData 的最新数据版本，而在 Observer#onChanged 中会对比两个版本号来决定是否分发。这个过程中没有使用反射，也不会存在不支持多观察者的问题。

public class ProtectedUnPeekLiveData<T> extends LiveData<T> {

    private final static int START_VERSION = -1;

    private final AtomicInteger mCurrentVersion = new AtomicInteger(START_VERSION);

    protected boolean isAllowNullValue;

    @Override
    public void observe(@NonNull LifecycleOwner owner, @NonNull Observer<? super T> observer) {
        super.observe(owner, createObserverWrapper(observer, mCurrentVersion.get()));
    }

    @Override
    public void observeForever(@NonNull Observer<? super T> observer) {
        super.observeForever(createObserverWrapper(observer, mCurrentVersion.get()));
    }

    public void observeSticky(@NonNull LifecycleOwner owner, @NonNull Observer<T> observer) {
        super.observe(owner, createObserverWrapper(observer, START_VERSION));
    }

    public void observeStickyForever(@NonNull Observer<? super T> observer) {
        super.observeForever(createObserverWrapper(observer, START_VERSION));
    }

    @Override
    protected void setValue(T value) {
        mCurrentVersion.getAndIncrement();
        super.setValue(value);
    }

    class ObserverWrapper implements Observer<T> {
        private final Observer<? super T> mObserver;
        private int mVersion = START_VERSION;

        public ObserverWrapper(@NonNull Observer<? super T> observer, int version) {
            this.mObserver = observer;
            this.mVersion = version;
        }

        @Override
        public void onChanged(T t) {
            if (mCurrentVersion.get() > mVersion && (t != null || isAllowNullValue)) {
                mObserver.onChanged(t);
            }
        }

        @SuppressWarnings("unchecked")
        @Override
        public boolean equals(Object o) {
            if (this == o) {
                return true;
            }
            if (o == null || getClass() != o.getClass()) {
                return false;
            }
            ObserverWrapper that = (ObserverWrapper) o;
            return Objects.equals(mObserver, that.mObserver);
        }

        @Override
        public int hashCode() {
            return Objects.hash(mObserver);
        }
    }

    @Override
    public void removeObserver(@NonNull Observer<? super T> observer) {
        if (observer.getClass().isAssignableFrom(ObserverWrapper.class)) {
            super.removeObserver(observer);
        } else {
            super.removeObserver(createObserverWrapper(observer, START_VERSION));
        }
    }

    private ObserverWrapper createObserverWrapper(@NonNull Observer<? super T> observer, int version) {
        return new ObserverWrapper(observer, version);
    }

    public void clear() {
        super.setValue(null);
    }
}


public class ProtectedUnPeekLiveData<T> extends LiveData<T> {

    private final static int START_VERSION = -1;

    private final AtomicInteger mCurrentVersion = new AtomicInteger(START_VERSION);

    protected boolean isAllowNullValue;

    @Override
    public void observe(@NonNull LifecycleOwner owner, @NonNull Observer<? super T> observer) {
        super.observe(owner, createObserverWrapper(observer, mCurrentVersion.get()));
    }

    @Override
    public void observeForever(@NonNull Observer<? super T> observer) {
        super.observeForever(createObserverWrapper(observer, mCurrentVersion.get()));
    }

    public void observeSticky(@NonNull LifecycleOwner owner, @NonNull Observer<T> observer) {
        super.observe(owner, createObserverWrapper(observer, START_VERSION));
    }

    public void observeStickyForever(@NonNull Observer<? super T> observer) {
        super.observeForever(createObserverWrapper(observer, START_VERSION));
    }

    @Override
    protected void setValue(T value) {
        mCurrentVersion.getAndIncrement();
        super.setValue(value);
    }

    class ObserverWrapper implements Observer<T> {
        private final Observer<? super T> mObserver;
        private int mVersion = START_VERSION;

        public ObserverWrapper(@NonNull Observer<? super T> observer, int version) {
            this.mObserver = observer;
            this.mVersion = version;
        }

        @Override
        public void onChanged(T t) {
            if (mCurrentVersion.get() > mVersion && (t != null || isAllowNullValue)) {
                mObserver.onChanged(t);
            }
        }

        @SuppressWarnings("unchecked")
        @Override
        public boolean equals(Object o) {
            if (this == o) {
                return true;
            }
            if (o == null || getClass() != o.getClass()) {
                return false;
            }
            ObserverWrapper that = (ObserverWrapper) o;
            return Objects.equals(mObserver, that.mObserver);
        }

        @Override
        public int hashCode() {
            return Objects.hash(mObserver);
        }
    }

    @Override
    public void removeObserver(@NonNull Observer<? super T> observer) {
        if (observer.getClass().isAssignableFrom(ObserverWrapper.class)) {
            super.removeObserver(observer);
        } else {
            super.removeObserver(createObserverWrapper(observer, START_VERSION));
        }
    }

    private ObserverWrapper createObserverWrapper(@NonNull Observer<? super T> observer, int version) {
        return new ObserverWrapper(observer, version);
    }

    public void clear() {
        super.setValue(null);
    }
}

# 随笔