LiveData 数据倒灌：别问，问就是不可预期 - Perfect alternative to SingleLiveEvent, supporting multiple observers.

你好，我在阅读源码试着理解版本代码演进的逻辑，但是我发现 V5版本还是存在HashMap恒久存在，注册的 Observer 越多，占用的内存越大问题

发现在退出页面调用removeObserver 方法时，入参 observer 进行 Hash 运算得到的 observeKey 与注册 Observer 时得到的observeKey 不一样

使用 getApplicationScopeViewModel(Class modelClass) 方法初始化 ViewModel 时，如果传入的是一个 AndroidViewModel,会报 java.lang.Class<com.kunminx.puremusic.domain.message.SharedViewModel> has no zero argument constructor

想请教下 AndroidViewModel 怎么使用 Application context 初始化？

描述问题

我是一个 MVVM 架构和响应式编程深度使用的一个开发者. Github 上刷到你的这个仓库. 我还是想对你们所说的数据倒灌进行解释说明一下.

你所说的数据倒灌其实根本不是一个问题或者 bug. LiveData 设计就是如此. 接受最近一个信号. 对应流的 Behavior 模式. 我们有知名度一点的流的实现有 RxJava 和 Kotlin 的 Flow. 在他们的实现中, 分别对应 BehaviorSubject 和 StateFlow 他们的图示如下, 你们可以看到, 在不同的时间点发生订阅, 你总是能收到最近的一个信号. 除非一开始就没发射过信号. 而 LiveData 正是类似于此种模式. 所以你们说它的数据倒灌, 其实根本不是问题, 人家设计本是如此.

那你们说的数据倒灌根本原因是因为什么呢? 其实是因为你们监听了 Behavior 模式的流或者 LiveData 去做了相应的操作. 比如你监听一个 LiveData 去做了网络请求 当你界面第一次进入, LiveData 中产生了一个信号, 你收到之后做了一次请求, 后来由于系统配置更改引起界面重建, 但是 ViewModel 还是原先那个, 所以在界面重建后你去监听 LiveData, 就会立马收到一个信号, 导致你又做了一次请求. 这里说明的场景, 是由于错误使用了 LiveData 引起的. 如果你要监听一个信号做一定的行为, 这类通常是需要监听 Publish 模式的流. 而 LiveData 设计之初就是 Behavior 模式, Publish 模式的行为示意图如下, 你只能收到你订阅点之后的信号.

总结

• Behavior 模式可选的方案
  • RxJava 的 BehaviorSubject
  • Kotlin Flow 的 StateFlow
  • LiveData
• Publish 模式可选的方案
  • RxJava 的 PublishSubject
  • Kotlin Flow 的 SharedFlow
  • 我们自定义的 Listener 等

综上所述, 在响应式编程中, 由于你接受的信号源有不同的模式实现. 所以在平常的业务需求中, 我们也要合理的进行选择.

比如我们用于显示界面的场景, Behavior 模式是最适合不过了, 这也是为什么 LiveData 出现的原因. 本就为了显示页面的数据去的. 在界面重建也能重新进行显示.

再比如我们用于执行某些行为的场景, 比如你收到一个信号进行一次网络请求、数据库操作、跳转等等. 这些其实都是需要使用 Publish 模式的.

希望我在这里的长篇大论, 能很好的解释你们出现的所谓的数据倒灌的问题. 在项目中能合理的选择对应的实现去解决问题. 并且对一个响应式的数据源进行监听的时候, 需要先知道它的实现模式是 Behavior 还是 Publish, 以便于你做出判断, 可以用作哪些场景的使用

只能在ViewModel中写数据。

private val _phoneInput: MutableLiveData<String>

val phoneInput: LiveData<String>
        get() = _phoneInput

外部只能拿到phoneInput，只能observe。 希望UnPeekLiveData也能做相应操作。

倒灌问题的产生原因：1.重新注册监听，2.livedata有数据。如果livedata用来发送事件场景时，应该是只关心添加监听之后的数据，添加之前的数据是不关心的。 倒灌问题的解决方案的思考：1.数据的管理者可能不关心数据的消费者是要什么样的数据（粘性、还是非粘性），2.livedata 的行为结果整个应用内应该是一致。因此我认为扩展方法比修改原方法可能更合适。 不成熟的解决方案：

/**
 * LiveData扩展事件监听,去掉粘性,去掉空数据
 * 订阅之前的数据不会回调。只会收到非空的数据。只在前台是收到回调,自动反注册
 */
fun <T> LiveData<T>.observeEvent(owner : LifecycleOwner,observer : Observer<T>){
    val ob = ObserverWrapper(this,observer)
    this.observe(owner,ob)
}

/**
 * 包装一下，用于判断是不是添加监听之后的数据,之前的数据不回调,空数据不回调
 * 事件不能为空
 */
internal class ObserverWrapper<T>(liveData: LiveData<T>,private val mObserver: Observer<in T>) : Observer<T> {
    private var skipValue = liveData.value !=null
    override fun onChanged(t: T) {
        if (!skipValue && t != null) {
            mObserver.onChanged(t)
        }
        skipValue = false
    }

    override fun equals(o: Any?): Boolean {
        if (this === o) {
            return true
        }
        if (o == null || javaClass != o.javaClass) {
            return false
        }
        val that = o as ObserverWrapper<*>
        return mObserver == that.mObserver
    }

    override fun hashCode(): Int {
        return Objects.hash(mObserver)
    }
}

open class ProtectedUnPeekLiveData<T : Any?> : LiveData<T?>() {

    private val pending = AtomicBoolean(false)
    protected var isAllowNullValue = false

    override fun observe(owner: LifecycleOwner, observer: Observer<in T?>) {
        super.observe(owner) { t ->
            if (pending.compareAndSet(true, false)
                && (t != null || isAllowNullValue)
            ) {
                observer.onChanged(t)
            }
        }
    }

    override fun observeForever(observer: Observer<in T?>) {
        super.observeForever { t ->
            if (pending.compareAndSet(true, false)
                && (t != null || isAllowNullValue)
            ) {
                observer.onChanged(t)
            }
        }
    }

    fun observeSticky(owner: LifecycleOwner, observer: Observer<T?>) {
        super.observe(owner, observer)
    }

    fun observeStickyForever(observer: Observer<T?>) {
        super.observeForever(observer)
    }

    override fun setValue(value: T?) {
        if (value != null || isAllowNullValue) {
            pending.set(true)
            super.setValue(value)
        }
    }

    fun clear() {
        super.setValue(null)
    }
}

public UnPeekLiveData create() { UnPeekLiveData liveData = new UnPeekLiveData<>(); liveData.isAllowNullValue = this.isAllowNullValue; return liveData; }

UnPeekLiveData 没有这个isAllowNullValue 变量。

重构 UnPeekLiveDataV4版本

最终实现对谷歌原生 LiveData 完全无侵入性的目的。在多人协作的场景下，其他同学就只需要理解 UnPeekLiveData 能够解决粘性事件、数据倒灌问题，其他的完全不需要理解，用法上完全跟原生 LiveData 保持一致

问题一：UnPeekLiveData 中的 observers 这个 HashMap 恒久存在，注册的 Observer 越多，占用的内存越大，并且除非UnPeekLiveData 被回收，否则恒久存在内存当中

解决：在 removeObserver 方法中移除 map 中对应存储的 storeId

问题二：无法通过 removeObserver 方法移除指定的 Observer（某些场景需要提前 removeObserver）

解决：通过维护外部传入 Observer 与内部代理 Observer 的映射关系，在 removeObserver 调用时，通过反射找到真正注册到 LiveData 中的 Observer，实现移除

问题三：同一个 Observer 对象，注册多次，UnPeekLiveData 内部实际上会注册了多个不同的 Observer，从而导致重复回调，产生一些不可预期的问题

解决：内部不会每次调用 observe 方法时都新创建一个代理 Observer，而是复用已经存在的代理 Observer（注意！！！Kotlin + LiveData + Lamda 由于编译器优化，可能会抛 Cannot add the same observer with different lifecycles 异常）

问题四：无法使用 observerForever 方法

解决：UnPeekLiveData 内部直接持有 forever 类型的 Observer

当使用 removeObservers(@NonNull final LifecycleOwner owner) 的时候，回调过来已经是ObserverWrapper，又封装了一层，导致没有删除掉。应该是加一个判断，判断回调当前类是否已经是ObserverWrapper的类型，不是才需要包装删除，那时候没有测试想到。 @Override public void removeObserver(@NonNull @NotNull Observer<? super T> observer) { if (observer.getClass().isAssignableFrom(ObserverWrapper.class)) { super.removeObserver(observer); } else { super.removeObserver(createObserverWrapper(observer, START_VERSION, false)); } }

public class EventMutableLiveData extends LiveData {

private final AtomicInteger currentVersion = new AtomicInteger(-1);

/**
 * 这个标志主要是是否忽略最新值， observe(@NonNull @NotNull LifecycleOwner owner, @NonNull @NotNull Observer<? super T> observer) 才会生效
 * 比如从页面1 来到页面2 页面2 再回到页面1 页面1默认是会收到页面2 最新的值，不想收到这个值可以设置为true
 */
private boolean isIgnoreLastObserve;

private final HashMap<Observer<? super T>, ObserverProxy> observerMap = new HashMap<>();

public EventMutableLiveData(boolean isIgnoreLastObserve) {
    super();
    this.isIgnoreLastObserve = isIgnoreLastObserve;
}

public EventMutableLiveData() {
    super();
}

@Override
public void observe(@NonNull @NotNull LifecycleOwner owner, @NonNull @NotNull Observer<? super T> observer) {
    if (!observerMap.containsKey(observer)) {
        observerMap.put(observer, new ObserverProxy(observer, currentVersion.get(), false));
    } else {
        Objects.requireNonNull(observerMap.get(observer)).changeVersion(currentVersion.get());
    }
    super.observe(owner, Objects.requireNonNull(observerMap.get(observer)));
}

@Override
public void observeForever(@NonNull @NotNull Observer<? super T> observer) {
    if (!observerMap.containsKey(observer)) {
        observerMap.put(observer, new ObserverProxy(observer, currentVersion.get(), true));
    }
    super.observeForever(Objects.requireNonNull(observerMap.get(observer)));
}

@Override
public void setValue(T value) {
    currentVersion.getAndIncrement();
    super.setValue(value);
}

@Override
public void postValue(T value) {
    currentVersion.getAndIncrement();
    super.postValue(value);
}

class ObserverProxy implements Observer<T> {

    private final Observer<? super T> observer;

    private int mVersion = -1;

    private final boolean isForever;

    public ObserverProxy(@NotNull Observer<? super T> observer, int mVersion, boolean isForever) {
        this.observer = observer;
        this.mVersion = mVersion;
        this.isForever = isForever;
    }

    @Override
    public synchronized void onChanged(T t) {
        if (currentVersion.get() > mVersion) {
            observer.onChanged(t);
            if (isIgnoreLastObserve && !isForever) {
                Set<Map.Entry<Observer<? super T>, ObserverProxy>> set = observerMap.entrySet();
                for (Map.Entry<Observer<? super T>, ObserverProxy> item : set) {
                    item.getValue().changeVersion(currentVersion.get());
                }
            }
        }
    }

    private void changeVersion(int newVersion) {
        this.mVersion = newVersion;
    }
}

@Override
public void removeObserver(@NonNull @NotNull Observer<? super T> observer) {
    super.removeObserver(observer);
    observerMap.remove(observer);
}

public void removeAllObserver() {
    Set<Map.Entry<Observer<? super T>, ObserverProxy>> set = observerMap.entrySet();
    for (Map.Entry<Observer<? super T>, ObserverProxy> item : set) {
        super.removeObserver(item.getValue());
    }
    observerMap.clear();
}

}

observerStateMap 这个map 感觉没有存在的必要,可以更加精简

private final ConcurrentHashMap<Observer<? super T>, ChangeablePair<Observer<? super T>, Boolean>> observerProxyMap = new ConcurrentHashMap();

这是我的想法,一个map就可以搞定

我觉得数据上会更好

另外针对这一行代码 改造成如下:

如有 bug，请另外 new 一个 issue ⚠️⚠️⚠️

本项目开 issue 规范：

1. 有任何 bug 都欢迎及时开 issue，我看到后予以处理。
2. 如有使用上的疑问，请先认真阅读 Readme 和源码 sample，在没有找到答案后，另外开 issue。
3. 如开 issue 是为了发表个人见解，请务必 客观、具体、严谨；严禁草率、乱入、带节奏：

务必注明观点所对应的场景，并附上完整可复现的代码，

不然缺乏一致的前提依据来有效交流！

任何缺乏实证依据和因果逻辑的泛泛而谈，都可能对其他使用者造成困扰。

在发表个人见解前，请先确保自己认真阅读过源码。这是对自己、对作者、对其他读者最起码的尊重。