Jetpack MVVM 中正确处理 Events
前言
在 MVVM 架构中，我们通常使用 LiveData 或者 StateFlow 实现 ViewModel 与 View 之间的数据通信，它们具备的响应式机制非常适合用来向 UI 侧发送更新后的状态（State），但是同样用它们来发送事件（Event），当做 EventBus 使用就不妥了
“状态” 与 “事件”
虽然“状态”和“事件”都可以通过响应式的方式通知到 UI 侧，但是它们的消费场景不同：
状态（State）：是需要 UI 长久呈现的内容，在新的状态到来之前呈现的内容保持不变。比如显示一个Loading框或是显示一组请求的数据集。 事件（Event）：是需要 UI 即时执行的动作，是一个短期行为。比如显示一个 Toast 、 SnackBar，或者完成一次页面导航等。
我们从覆盖性、时效性、幂等性等三个维度列举状态和事件的具体区别
| 状态 | 事件 | |
|---|---|---|
| 覆盖性 | 新状态会覆盖旧状态，如果短时间内发生多次状态更新，可以抛弃中间态只保留最新状态即可。这也是为什么 LiveData 连续 postValue 时会出现数据丢失。 | 新事件不应该覆盖旧事件，订阅者按照发送顺序接收到所有事件，中间的事件不能遗漏。 |
| 时效性 | 最新状态是需要长久保持的，可以被时刻访问到，因此状态一般是“粘性的”，在新的订阅出现时为其发送最新状态。 | 事件只能被消费一次，消费后应该丢弃。因此事件一般不是“粘性”的，避免多次消费。 |
| 幂等性 | 状态是幂等的，唯一状态决定唯一UI，同样的状态无需响应多次。因此 StateFlow 在 setValue 时会对新旧数据进行比较，避免重复发送。 | 订阅者需要对发送的每个事件进行消费，即使是同一类事件发送多次。 |
基于 LiveData 的事件处理
鉴于事件与状态的诸多差异，如果直接使用 LiveData 或 StateFlow 发送事件，会出现不符合预期的行为。其中最常见的可能就是所谓“数据倒灌”问题。
我平常不太使用 “数据倒灌” 这个词，主要是 “倒”这个字与单向数据流思想相违背，容易引起误解，我猜测词汇发明者更多的是想强调一种 “被动”接受吧。
“数据倒灌”问题的发生源于 LiveData 的 "粘性" 设计，同一个订阅者每次订阅 LiveData 都会收到最近的一个事件，因为事件应该具有“时效性”，对于已消费过的事件我们不希望再次响应。
Jose Alcérreca 在 《LiveData with SnackBar, Navigation and other events》[1] 一文中首次讨论了 LiveData 如何处理事件的话题，并在 architecture-sample-todoapp 中给出了 SingleLiveEvent 的解决思路。受到这篇文章的启发，陆续又有不少大佬给出了更优的解决方案，修补了 SingleLiveEvent 中的一些缺陷 - 例如不支持多订阅者等，但主要的解决思路上大体相同：通过增加标记位来记录事件是否被消费，对于已消费的事件则不会在订阅时再次发送。
这里贴一个相对完善的解决方案：
open class LiveEvent<T> : MediatorLiveData<T>() {
private val observers = ArraySet<ObserverWrapper<in T>>()
@MainThread
override fun observe(owner: LifecycleOwner, observer: Observer<in T>) {
observers.find { it.observer === observer }?.let { _ -> // existing
return
}
val wrapper = ObserverWrapper(observer)
observers.add(wrapper)
super.observe(owner, wrapper)
}
@MainThread
override fun observeForever(observer: Observer<in T>) {
observers.find { it.observer === observer }?.let { _ -> // existing
return
}
val wrapper = ObserverWrapper(observer)
observers.add(wrapper)
super.observeForever(wrapper)
}
@MainThread
override fun removeObserver(observer: Observer<in T>) {
if (observer is ObserverWrapper && observers.remove(observer)) {
super.removeObserver(observer)
return
}
val iterator = observers.iterator()
while (iterator.hasNext()) {
val wrapper = iterator.next()
if (wrapper.observer == observer) {
iterator.remove()
super.removeObserver(wrapper)
break
}
}
}
@MainThread
override fun setValue(t: T?) {
observers.forEach { it.newValue() }
super.setValue(t)
}
private class ObserverWrapper<T>(val observer: Observer<T>) : Observer<T> {
private var pending = false
override fun onChanged(t: T?) {
if (pending) {
pending = false
observer.onChanged(t)
}
}
fun newValue() {
pending = true
}
}
}
代码很清晰，我们使用 ObserverWrapper 对 Observer 进行封装后，可以使用 pending 针对单个消费者记录事件的消费，避免二次消费。
简单介绍了 LiveData 的事件处理，接下来重点看一下 Flow 如何进行事件处理，因为随着 lifecycle-runtime-ktx 对 Coroutine 的支持， Flow 将会成为主流的数据通信方式，Flow 将会成为主流的数据通信方式。
基于 SharedFlow 的事件处理
StateFlow 和 LiveData 一样具备“粘性”特性，同样有“数据倒灌”的问题，甚至更有过之还会出现“数据丢失”的问题，因为 StateFlow 进行 updateState 时会过滤对新旧数据进行比较，同样类型的事件有可能被丢弃。
Roman Elizarov 曾在 《Shared flows, broadcast channels》 一文中提出用 SharedFlow 实现 EventBus 的做法：
class BroadcastEventBus {
private val _events = MutableSharedFlow<Event>()
val events = _events.asSharedFlow() // read-only public view
suspend fun postEvent(event: Event) {
_events.emit(event)
}
}
SharedFlow 确实一个不错的选择，它的很多特性与事件消费方式比较贴合:
首先，它可以有多个收集器（订阅者），多个收集器“共享”事件，实现事件的广播，如下图所示：
其次，SharedFlow 的数据会以流的形式发送，不会丢失，新事件不会覆盖旧事件； 最后，它的数据不是粘性的，消费一次就不会再次出现。
但是，SharedFlow 存在一个问题，接收器无法接收到 collect 之前发送的事件，看下面例子：
class MainViewModel : ViewModel(), DefaultLifecycleObserver {
private val _toast = MutableSharedFlow<String>()
val showToast = _toast.asSharedFlow()
init {
viewModelScope.launch {
delay(1000)
_toast.emit("Toast")
}
}
}
//Fragment side
viewLifecycleOwner.lifecycleScope.launch {
viewLifecycleOwner.repeatOnLifecycle(Lifecycle.State.STARTED) {
mainViewModel.showToast.collect {
Toast.makeText(context, it, Toast.LENGTH_SHORT).show()
}
}
}
例子中，我们使用 repeatOnLifecycle 保证了事件收集在 STARTD 之后开始，如果此时注释掉 delay(1000) 的代码，emit 早于 collect，所以 toast 将无法显示。
有些时候我们在订阅出现之前就发出事件，并希望订阅者出现时执行响应这个事件，比如完成一个初始化任务等，注意这并非一种“数据倒灌”，因为这它只被允许消费一次，一旦消费就不再发送，所以 SharedFlow 的 replay 参数不能使用，因为 repaly 不能保证只消费一次。
基于 Channel 的处理事件
针对 SharedFlow 的这个不足， Roman Elizarov 也给了解决方案，即使用 Channel。
class SingleShotEventBus {
private val _events = Channel<Event>()
val events = _events.receiveAsFlow() // expose as flow
suspend fun postEvent(event: Event) {
_events.send(event) // suspends on buffer overflow
}
}
当 Channel 没有订阅者时，向其发送的数据会挂起，保证订阅者出现时第一时间接收到这个数据，类似于阻塞队列的原理。Channel 本身也是 Flow 实现的基础，所以通过 receiveAsFlow 可以转成一个 Flow 暴露给订阅者。回看前面的例子，改为 Channel 后如下：
class MainViewModel : ViewModel(), DefaultLifecycleObserver {
private val _toast = Channel<String>()
val showToast = _toast.receiveAsFlow()
init {
viewModelScope.launch {
_toast.send("Toast")
}
}
}
//Fragment side
viewLifecycleOwner.lifecycleScope.launch {
viewLifecycleOwner.repeatOnLifecycle(Lifecycle.State.STARTED) {
mainViewModel.showToast.collect {
Toast.makeText(context, it, Toast.LENGTH_SHORT).show()
}
}
}
UI 侧仍然针对 Flow 订阅，代码不做任何改动，但是在 STATED 之后也可以接受到已发送的事件。
需要注意，Channel 也有一个使用上的限制，当 Channel 有多个收集器时，它们不能共享 Channel 传输的数据，每个数据只能被一个收集器独享，因此 Channel 更适合一对一的通信场景。
综上，SharedFlow 和 Channel 在事件处理上各有特点，大家需要根据实际场景灵活选择：
| SharedFlow | Channel | |
|---|---|---|
| 订阅者数量 | 订阅者共享通知，可以实现一对多的广播 | 每个消息只有一个订阅者可以收到，用于一对一的通信 |
| 事件接受 | collect 之前的事件会丢失 | 第一个订阅者可以收到 collect 之前的事件 |
为了在更正确的时机接受事件，通常会配合 lifecycle-runtime-ktx 完成事件订阅，例如前面例子中使用的 repeatOnLifecycle ，这里提供一个避免模板代码的方法，仅供参考
inline fun <reified T> Flow<T>.observeWithLifecycle(
lifecycleOwner: LifecycleOwner,
minActiveState: Lifecycle.State = Lifecycle.State.STARTED,
noinline action: suspend (T) -> Unit
): Job = lifecycleOwner.lifecycleScope.launch {
flowWithLifecycle(lifecycleOwner.lifecycle, minActiveState).collect(action)
}
inline fun <reified T> Flow<T>.observeWithLifecycle(
fragment: Fragment,
minActiveState: Lifecycle.State = Lifecycle.State.STARTED,
noinline action: suspend (T) -> Unit
): Job = fragment.viewLifecycleOwner.lifecycleScope.launch {
flowWithLifecycle(fragment.viewLifecycleOwner.lifecycle, minActiveState).collect(action)
}
如上，observeWithLifecycle 作为 Flow 的扩展方法，在指定生命周期进行订阅，这样在 UI 侧的代码可以简写如下了：
viewModel.events
.observeWithLifecycle(fragment = this, minActiveState = Lifecycle.State.RESUMED) {
// do things
}
viewModel.events
.observeWithLifecycle(lifecycleOwner = viewLifecycleOwner, minActiveState = Lifecycle.State.RESUMED) {
// do things
}
本来文章到这里就该结束了，但突然发现近日 Google 对架构规范进行了更新，其中特别对 MVVM 的事件处理给了新的推荐做法：https://developer.android.com/jetpack/guide/ui-layer/events#handle-viewmodel-events，因此又有了下面一节内容......
关于 Google 最新 Guide
这里仅针对 Guide 中关于事件处理部分做一个摘要，可以总结为以下三条：
凡是发送给 View 的事件都应该涉及 UI 变动，与 UI 无关的事件不应该由 View 监听 既然是涉及 UI 的事件，可以跟随 UI 状态一起发送（基于 StateFlow 或 LiveData ），不必另建新的渠道。 View 在处理完事件后，需要告知 ViewModel 事件已处理，ViewModel 更新状态避免再次消费
这三条汇总成一句话就是：像 “状态” 一样管理 “事件”
结合官方的实例代码，体会一下具体实现：
// Models the message to show on the screen.
data class UserMessage(val id: Long, val message: String)
// Models the UI state for the Latest news screen.
data class LatestNewsUiState(
val news: List<News> = emptyList(),
val isLoading: Boolean = false,
val userMessages: List<UserMessage> = emptyList()
)
如上，List<UserMessge> 作为消息事件列表，跟 UiState 放在一起管理。
class LatestNewsViewModel(/* ... */) : ViewModel() {
private val _uiState = MutableStateFlow(LatestNewsUiState(isLoading = true))
val uiState: StateFlow<LatestNewsUiState> = _uiState
fun refreshNews() {
viewModelScope.launch {
// If there isn't internet connection, show a new message on the screen.
if (!internetConnection()) {
_uiState.update { currentUiState ->
val messages = currentUiState.userMessages + UserMessage(
id = UUID.randomUUID().mostSignificantBits,
message = "No Internet connection"
)
currentUiState.copy(userMessages = messages)
}
return@launch
}
// Do something else.
}
}
}
如上，ViewModel 在 refreshNews 中请求最新的数据，如果网络未连接，则增加一条 userMessage 跟随状态一起发送给 View 。
class LatestNewsActivity : AppCompatActivity() {
private val viewModel: LatestNewsViewModel by viewModels()
override fun onCreate(savedInstanceState: Bundle?) {
/* ... */
lifecycleScope.launch {
repeatOnLifecycle(Lifecycle.State.STARTED) {
viewModel.uiState.collect { uiState ->
uiState.userMessages.firstOrNull()?.let { userMessage ->
// TODO: Show Snackbar with userMessage.
// Once the message is displayed and
// dismissed, notify the ViewModel.
viewModel.userMessageShown(userMessage.id)
}
...
}
}
}
}
}
View 侧订阅 UiState 的状态变化，收到状态变化通知时，处理其中的 UserMessage 事件，例如这里是显示一条 SnackBar ，事件处理后，调用 viewModel.userMessageShown 方法，通知 ViewModel 处理结束。
fun userMessageShown(messageId: Long) {
_uiState.update { currentUiState ->
val messages = currentUiState.userMessages.filterNot { it.id == messageId }
currentUiState.copy(userMessages = messages)
}
}
最后看一下 userMessageShown 的实现，从消息列表中删除相关信息，表示消息已被消费。
其实 Jose Alcérreca 早在 《LiveData with SnackBar, Navigation and other events》[2] 一文中就提到过这种处理思路，并予以了否定，
With this approach you add a way to indicate from the View that you already handled the event and that it should be reset.
The problem with this approach is that there’s some boilerplate (one new method in the ViewModel per event) and it’s error prone; it’s easy to forget the call to the ViewModel from the observer.
否定的理由是这会增加模板代码，而且容易遗漏 View -> ViewModel 的反向通知。虽说 Jose 的文章只代表个人，但由于文章已经深入人心，如今 Google 的反向推荐难免让人感觉有些打脸。不过细细想来，这种做法也确实有它的意义：
它避免了 SharedFlow ，Channel 等更多工具的引入，技术栈更加简洁。 弱化 “事件” 的概念，强化 “状态” 的概念，实则就是命令式逻辑为状态驱动的思考方式让路，这也与 Compose 的理念更加贴近，有利于声明式 UI 的进一步推广 像 “状态” 一样管理 “事件”，事件处理有回执、可追踪，也为事件增加了“后处理”的机会
当然这里也存在隐患，比如在事件处理结束并给出回执之前，如果有新的状态通知到来，此时由于事件列表中没有清空当前事件，是否会造成重复消费？这个还有待进一步验证。
总结
本文介绍了 MVVM 事件处理的多种方案，没有十全十美的方案，需要大家结合具体场景做出选择：
如果你的项目仍然在使用 LiveData，那么需要对事件的消费做记录，避免事件二次消费，可以参考本文中 LiveEvent 的例子 如果你的代码大部分是 Kotlin ，那么推荐优先使用 Coroutine 实现 MVVM 数据通信，此时可以使用 SharedFlow 处理事件，如果你希望接收到 collect 之前的事件则可以选择 Channel 有条件的话可以考虑采用 Google 最新的架构规范，虽然它在写法上略显冗余，而且增加了 View 的负担，所以能否得到开发者的最终认可还有待检验。
其实最有效的事件处理方式就是尽量避免定义 “事件”，尝试用 “状态” 代替 “事件” 来设计你的数据通信 ，这才更贴合数据驱动的架构思想。
参考
- [1] Jose Alcérreca , https://medium.com/androiddevelopers/livedata-with-snackbar-navigation-and-other-events-the-singleliveevent-case-ac2622673150
- [2] Roman Elizarov , https://elizarov.medium.com/shared-flows-broadcast-channels-899b675e805c
推荐文章
加好友进交流群，技术干货聊不停