引子

在上一篇中，用 MVI 重构了“新闻流”这个业务场景。本篇在此基础上进一步拓展，引入 MVI 中两个重要的概念PartialChange和Reducer。

假设“新闻流”这个业务场景，用户可以触发如下行为：

1. 初始化新闻流
2. 上拉加载更多新闻
3. 举报某条新闻

在 MVVM 中，这些行为被表达为 ViewModel 的一个方法调用。在 MVI 中被称为意图Intent，它们不再是一个方法调用，而是一个数据。通常可被这样定义：

sealed class FeedsIntent {
    data class Init(val type: Int, val count: Int) : FeedsIntent()
    data class More(val timestamp: Long, val count: Int) : FeedsIntent()
    data class Report(val id: Long) : FeedsIntent()
}

这样做使得界面意图都以数据的形式流入到一个流中，好处是，可以用流的方式统一管理所有意图。
产品文档定义了所有的用户意图Intent，而设计稿定义了所有的界面状态State：

data class NewsState(
    val data: List<News>, // 新闻列表
    val isLoading: Boolean, // 是否正在首次加载
    val isLoadingMore: Boolean, // 是否正在上拉加载更多
    val errorMessage: String, // 加载错误信息 toast
    val reportToast: String, // 举报结果 toast
) {
    companion object {
        // 新闻流的初始状态
        val initial = NewsState(
            data = emptyList(), 
            isLoading = true, 
            isLoadingMore = false, 
            errorMessage = "",
            reportToast = ""
        )
    }
}

在 MVI 中，把界面的一次展示理解为单个 State 的一次渲染。相较于 MVVM 中一个界面可能被分拆为多个 LiveData，State 这种唯一数据源降低了复杂度，使得代码容易维护。

有了 Intent 和 State，整个界面刷新的过程就形成了一条单向数据流，如下图所示：

MVI 就是用“响应式编程”的方式将这条数据流中的若干 Intent 转换成唯一 State。初级的转换方式是直接将 Intent 映射成 State。

PartialChange

理论上 Intent 是无法直接转换为 State 的。因为 Intent 只表达了用户触发的行为，而行为产生的结果才对应一个 State。更具体的说，“上拉加载更多新闻”可能产生三个结果：

1. 正在加载更多新闻。
2. 加载更多新闻成功。
3. 加载更多新闻失败。

其中每一个结果都对应一个 State。“单向数据流”内部的数据变换详情如下：

每一个意图会产生若干个结果，每个结果对应一个界面状态。

上图看着有“很多条”数据流，但同一时间只可能有一条起作用。上图看着会在 ViewModel 内部形成各种 State，但暴露给界面的还是唯一 State。

因为所有意图产生的所有可能的结果都对应于一个唯一 State 实例，所以每个意图产生的结果只引起 State 部分字段的变化。比如 Init.Success 只会影响 NewsState.data 和 NewsState.isLoading。

在 MVI 框架中，意图 Intent 产生的结果称为部分变化PartialChange。

总结一下：

• MVI 框架中用数据流来理解界面刷新。

• 数据流的起点是界面发出的意图（Intent），一个意图会产生若干结果，它们称为 PartialChange，一个 PartialChange 对应一个 State 实例。

• 数据流的终点是界面对 State 的观察而进行的一次渲染。

连续的状态

界面展示的变化是“连续的”，即界面新状态总是由上一次状态变化而来。就像连环画一样，下一帧是基于上一帧的偏移量。

这种基于老状态产生新状态的行为称为Reduce，用一个 lambda 表达即是(oldState: State) -> State。

界面发出的不同意图会生成不同的结果，每种结果都有各自的方法进行新老状态的变换。比如“上拉加载更多新闻”和“举报新闻”，前者在老状态的尾部追加数据，而后者是在老状态中删除数据。

基于此，Reduce 的 lambda 可作如下表达：(oldState: State, change: PartialChange) -> State，即新状态由老状态和 PartialChange 共同决定。

通常 PartialChange 被定义成密封接口，而 Reduce 定义为内部方法：

// 新闻流的部分变化
sealed interface FeedsPartialChange {
    // 描述如何从老状态变化为新状态
    fun reduce(oldState: NewsState): NewsState
}

这是 PartialChange 的抽象定义，新闻流场景中，它应该有三个实现类，分别是 Init，More，Report。其中 Init 的实现如下：

sealed class Init : FeedsPartialChange {
    // 在初始化新闻流流场景下，老状态如何变化成新状态
    override fun reduce(oldState: NewsState): NewsState = 
        // 对初始化新闻流能产生的所有结果分类讨论，并基于老状态拷贝构建新状态
        when (this) {
            Loading -> oldState.copy(isLoading = true)
            is Success -> oldState.copy(
                data = news,//方便地访问Success携带的数据
                isLoading = false,
                isLoadingMore = false,
                errorMessage = ""
            )
            is Fail -> oldState.copy(
                data = emptyList(),
                isLoading = false,
                isLoadingMore = false,
                errorMessage = error
            )
    }
    // 加载中
    object Loading : Init()
    // 加载成功
    data class Success(val news: List<News>) : Init()
    // 加载失败
    data class Fail(val error: String) : Init()
}

初始化新闻流的 PartialChange 也被实现为密封的，密封产生的效果是，在编译时，其子类的全集就已经全部确定，不允许在运行时动态新增子类，且所有子类必须内聚在一个包名下。

这样做的好处是降低界面刷新的复杂度，即有限个 Intent 会产生有限个 PartialChange，且它们唯一对应一个 State。出 bug 的时候只需从三处找问题：1. Intent 是否发射？ 2. 是否生成了既定的 PartialChange？ 3. reduce 算法是否有问题？

将 reduce 算法定义在 PartialChange 内部，就能很方便地获取 PartialChange 携带的数据，并基于它构建新状态。

用同样的思路，More 和 Report 的定义如下：

sealed class More : FeedsPartialChange {
    override fun reduce(oldState: NewsState): NewsState = when (this) {
        Loading -> oldState.copy(
            isLoading = false,
            isLoadingMore = true,
            errorMessage = ""
        )
        is Success -> oldState.copy(
            data = oldState.data + news,// 新数据追加在老数据后
            isLoading = false,
            isLoadingMore = false,
            errorMessage = ""
        )
        is Fail -> oldState.copy(
            isLoadingMore = false,
            isLoading = false,
            errorMessage = error
        )
    }

    object Loading : More()
    data class Success(val news: List<News>) : More()
    data class Fail(val error: String) : More()
}

sealed class Report : FeedsPartialChange {
    override fun reduce(oldState: NewsState): NewsState = when (this) {
        is Success -> oldState.copy(
            // 在老数据中删除举报新闻
            data = oldState.data.filterNot { it.id == id },
            reportToast = "举报成功"
        )
        Fail -> oldState.copy(reportToast = "举报失败")
    }

    class Success(val id: Long) : Report()
    object Fail : Report()
}

状态的变换

Intent，PartialChange，Reduce，State 定义好了，是时候看看如何用流的方式把它们串联起来！

总体来说，状态是这样变换的：Intent -> PartialChange -(Reduce)-> State

1. Intent 流入，State 流出

class StateFlowActivity : AppCompatActivity() {
    private val newsViewModel by lazy {
        ViewModelProvider(
            this,
            NewsViewModelFactory(NewsRepo(this))
        )[NewsViewModel::class.java]
    }

    // 将所有意图通过 merge 进行合流
    private val intents by lazy {
        merge(
            flowOf(FeedsIntent.Init(1, 5)),// 初始化新闻
            loadMoreFlow(), // 加载更多新闻
            reportFlow()// 举报新闻
        )
    }

    // 将上拉加载更多转换成数据流
    private fun loadMoreFlow() = callbackFlow {
        recyclerView.setOnLoadMoreListener {
            trySend(FeedsIntent.More(111L, 2))
        }
        awaitClose { recyclerView.removeOnLoadMoreListener(null) }
    }

    // 将举报新闻转换成数据流
    private fun reportFlow() = callbackFlow {
        reportView.setOnClickListener {
            val news = newsAdapter.dataList[i] as? News
            news?.id?.let { trySend(FeedsIntent.Report(it)) }
        }
        awaitClose { reportView.setOnClickListener(null) }
    }

    override fun onCreate(savedInstanceState: Bundle?) {
        super.onCreate(savedInstanceState)
        setContentView(contentView)

        // 订阅意图流
        intents
            // Intent 流入 ViewModel
            .onEach(newsViewModel::send)
            .launchIn(lifecycleScope)
        // 订阅状态流
        newsViewModel.newState
            // State 流出 ViewModel，并绘制界面
            .collectIn(this) { showNews(it) }
    }
}

class NewsViewModel(private val newsRepo: NewsRepo) : ViewModel() {
    // 用于接收意图的 SharedFlow
    private val _feedsIntent = MutableSharedFlow<FeedsIntent>()
    // 意图被变换为状态
    val newState =
        _feedsIntent.map {} // 伪代码，省略了 将 Intent 变换为 State 的细节
    // 将意图发送到流
    fun send(intent: FeedsIntent) {
        viewModelScope.launch { _feedsIntent.emit(intent) }
    }
}

界面可以发出的所有意图都被组织到一个流中，并且罗列在一起。intents流可以作为理解业务逻辑的入口。同时 ViewModel 提供了一个 State 流，供界面订阅。

2. Intent -> PartialChange

class NewsViewModel(private val newsRepo: NewsRepo) : ViewModel() {
    private val _feedsIntent = MutableSharedFlow<FeedsIntent>()
    // 供界面观察的唯一状态
    val newState =
        _feedsIntent
            .toPartialChangeFlow()
            .flowOn(Dispatchers.IO)
            .stateIn(viewModelScope, SharingStarted.Eagerly,NewsState.initial)
    )
}

各种 Intent 转换为 PartialChange 的逻辑被封装在toPartialChangeFlow()中：

// NewsViewModel.kt
// 将 Intent 流变换为 PartialChange 流
private fun Flow<FeedsIntent>.toPartialChangeFlow(): Flow<FeedsPartialChange> = merge(
    // 过滤出初始化新闻意图并将其变换为对应的 PartialChange
    filterIsInstance<FeedsIntent.Init>().flatMapConcat { it.toPartialChangeFlow() },
    // 过滤出上拉加载更多意图并将其变换为对应的 PartialChange
    filterIsInstance<FeedsIntent.More>().flatMapConcat { it.toPartialChangeFlow() },
    // 过滤出举报新闻意图并将其变换为对应的 PartialChange
    filterIsInstance<FeedsIntent.Report>().flatMapConcat { it.toPartialChangeFlow() },
)

toPartialChangeFlow() 被定义为扩展方法。

filterIsInstance() 用于过滤出Flow<FeedsIntent>中的子类型并分类讨论，因为每种 Intent 变换为 PartialChange 的方式有所不同。

最后用 merge 进行合流，它会将每个 Flow 中的数据合起来并发地转发到一个新的流上。merge + filterIsInstance的组合相当于流中的 if-else。

其中的 toPartialChangeFlow() 是各种意图的扩展方法：

// NewsViewModel.kt
private fun FeedsIntent.Init.toPartialChangeFlow() =
    flowOf(
        // 本地数据库新闻
        newsRepo.localNewsOneShotFlow,
        // 网络新闻
        newsRepo.remoteNewsFlow(this.type.toString(), this.count.toString())
    )
        // 并发合流
        .flattenMerge()
        .transformWhile {
            emit(it.news)
            !it.abort
        }
        // 将新闻数据变换为成功或失败的 PartialChange
        .map { news -> 
            if (news.isEmpty()) Init.Fail("no news") else Init.Success(news) 
        }
        // 发射展示 Loading 的 PartialChange
        .onStart { emit(Init.Loading) }

该扩展方法描述了如何将 FeedsIntent.Init 变换为对应的 PartialChange。同样地，FeedsIntent.More 和 FeedsIntent.Report 的变换逻辑如下：

// NewsViewModel.kt
private fun FeedsIntent.More.toPartialChangeFlow() =
    newsRepo.remoteNewsFlow("news", "10")
        .map {news -> 
            if(it.news.isEmpty()) More.Fail("no more news") else More.Success(it.news) 
        }
        .onStart { emit(More.Loading) }
        .catch { emit(More.Fail("load more failed by xxx")) }

private fun FeedsIntent.Report.toPartialChangeFlow() =
    newsRepo.reportNews(id)
        .map { if(it >= 0L) Report.Success(it) else Report.Fail}
        .catch { emit((Report.Fail)) }

3. PartialChange -(Reduce)-> State

经过 toPartialChangeFlow() 的变换，现在流中流动的数据是各种类型的 PartialChange。接下来就要将其变换为 State：

// NewsViewModel.kt
val newState =
  _feedsIntent
    .toPartialChangeFlow()
    // 将 PartialChange 变换为 State
    .scan(NewsState.initial){oldState, partialChange -> partialChange.reduce(oldState)}
    .flowOn(Dispatchers.IO)
    .stateIn(viewModelScope, SharingStarted.Eagerly,NewsState.initial)
)

使用scan()进行变换：

// 从 Flow<T> 变换为 Flow<R>
public fun <T, R> Flow<T>.scan(
    initial: R, // 初始值
    operation: suspend (accumulator: R, value: T) -> R // 累加算法
): Flow<R> = runningFold(initial, operation)

public fun <T, R> Flow<T>.runningFold(
    initial: R, 
    operation: suspend (accumulator: R, value: T) -> R): Flow<R> = flow {
    // 累加器
    var accumulator: R = initial
    emit(accumulator)
    collect { value ->
        // 进行累加
        accumulator = operation(accumulator, value)
        // 向下游发射累加值
        emit(accumulator)
    }
}

从 scan() 的签名看，是将一个流变换为另一个流，看似和 map() 相似。但它的变换算法是带累加的。用 lambda 表达为(accumulator: R, value: T) -> R。

这不正好就是上面提到的 Reduce 吗！即基于老状态和新 PartialChange 生成新状态。

MVVM 和 MVI 复杂度比拼

就新闻流这个场景，用图来对比下 MVVM 和 MVI 复杂度的区别。

这张图表达了三种复杂度：

1. View 发起请求的复杂度：ViewModel 的各种方法调用会散落在界面不同地方。即界面向 ViewModel 发起请求没有统一入口。
2. View 观察数据的复杂度：界面需要观察多个 ViewModel 提供的数据，这导致界面状态的一致性难以维护。
3. ViewModel 内部请求和数据关系的复杂度：数据被定义为 ViewModel 的成员变量。成员变量是增加复杂度的利器，因为它可以被任何成员方法访问。也就是说，新增业务对成员变量的修改可能影响老业务的界面展示。同理，当界面展示出错时，也很难一下子定位到是哪个请求造成的。

再来看一下让人耳目一新的 MVI 吧：

完美化解上述三个没有必要的复杂度。

总之，用上 MVI 后，新需求不再破坏老逻辑，出 bug 了能更快速定位到问题。

敬请期待

还有一个问题有待解决，那就是 MVI 框架下，刷新界面时持久性状态 State 和 一次性事件 Event 的区别对待。

在 MVVM 中，因为 LiveData 的粘性，导致一次性事件被界面多次消费。对此有多种解决方案。

总结

• MVI 框架中用单向数据流来理解界面刷新。整个数据流中包含的数据依次如下：Intent，PartialChange，State

• 数据流的起点是界面发出的意图（Intent），一个意图会产生若干结果，它们称为 PartialChange，一个 PartialChange 对应一个 State 实例。

• 数据流的终点是界面对 State 的观察而进行的一次渲染。

• MVI 就是用“响应式编程”的方式将单向数据流中的若干 Intent 转换成唯一 State。

• MVI 强调的单向数据流表现在两个层面：

  1. View 和 ViewModel 交互过程中的单向数据流：单个Intent流流入 ViewModel，单个State流流出 ViewModel。
  2. ViewModel 内部数据变换的单向数据流：Intent 变换为多个 PartialChange，一个 PartialChange 对应一个 State。

Talk is cheap, show me the code

完整代码如下：

StateFlowActivity.kt

class StateFlowActivity : AppCompatActivity() {
    private val newsAdapter2 by lazy {
        VarietyAdapter2().apply {addProxy(NewsProxy())}
    }

    private val intents by lazy {
        merge(
            flowOf(FeedsIntent.Init(1, 5)),
            loadMoreFlow(),
            reportFlow()
        )
    }

    private fun loadMoreFlow() = callbackFlow {
        recyclerView.setOnLoadMoreListener {
            trySend(FeedsIntent.More(111L, 2))
        }
        awaitClose { recyclerView.removeOnLoadMoreListener(null) }
    }

    private fun reportFlow() = callbackFlow {
        reportView.setOnClickListener {
            val news = newsAdapter.dataList[i] as? News
            news?.id?.let { trySend(FeedsIntent.Report(it)) }
        }
        awaitClose { reportView.setOnClickListener(null) }
    }

    override fun onCreate(savedInstanceState: Bundle?) {
        super.onCreate(savedInstanceState)
        setContentView(contentView)

        intents
            .onEach(newsViewModel::send)
            .launchIn(lifecycleScope)

        newsViewModel.newState
            .collectIn(this) { showNews(it) }
    }

    private fun showNews(state: NewsState) {
        state.apply {
            if (isLoading) showLoading() else dismissLoading()
            if (isLoadingMore) showLoadingMore() else dismissLoadingMore()
            if (reportToast.isNotEmpty()) Toast.makeText(
                this@StateFlowActivity,
                state.reportToast,
                Toast.LENGTH_SHORT
            ).show()
            if (errorMessage.isNotEmpty()) tv.text = state.errorMessage
            if (data.isNotEmpty()) newsAdapter2.dataList = state.data
        }
    }
}

NewsViewModel.kt

class NewsViewModel(private val newsRepo: NewsRepo) : ViewModel() {
    private val _feedsIntent = MutableSharedFlow<FeedsIntent>()

    val newState =
        _feedsIntent
            .toPartialChangeFlow()
            .scan(NewsState.initial) { oldState, partialChange -> partialChange.reduce(oldState) }
            .flowOn(Dispatchers.IO)
            .stateIn(viewModelScope, SharingStarted.Eagerly,NewsState.initial)

    fun send(intent: FeedsIntent) {
        viewModelScope.launch { _feedsIntent.emit(intent) }
    }

    private fun Flow<FeedsIntent>.toPartialChangeFlow(): Flow<FeedsPartialChange> = merge(
        filterIsInstance<FeedsIntent.Init>().flatMapConcat { it.toPartialChangeFlow() },
        filterIsInstance<FeedsIntent.More>().flatMapConcat { it.toPartialChangeFlow() },
        filterIsInstance<FeedsIntent.Report>().flatMapConcat { it.toPartialChangeFlow() },
    )

    private fun FeedsIntent.More.toPartialChangeFlow() =
        newsRepo.remoteNewsFlow("", "10")
            .map { if (it.news.isEmpty()) More.Fail("no more news") else More.Success(it.news) }
            .onStart { emit(More.Loading) }
            .catch { emit(More.Fail("load more failed by xxx")) }

    private fun FeedsIntent.Init.toPartialChangeFlow() =
        flowOf(
            newsRepo.localNewsOneShotFlow,
            newsRepo.remoteNewsFlow(this.type.toString(), this.count.toString())
        )
            .flattenMerge()
            .transformWhile {
                emit(it.news)
                !it.abort
            }
            .map { news -> if (news.isEmpty()) Init.Fail("no more news") else Init.Success(news) }
            .onStart { emit(Init.Loading) }
            .catch {
                if (it is SSLHandshakeException)
                    emit(Init.Fail("network error,show old news"))
            }

    private fun FeedsIntent.Report.toPartialChangeFlow() =
        newsRepo.reportNews(id)
            .map { if(it >= 0L) Report.Success(it) else Report.Fail}
            .catch { emit((Report.Fail)) }
}

NewsState.kt

data class NewsState(
    val data: List<News> = emptyList(),
    val isLoading: Boolean = false,
    val isLoadingMore: Boolean = false,
    val errorMessage: String = "",
    val reportToast: String = "",
) {
    companion object {
        val initial = NewsState(isLoading = true)
    }
}

FeedsPartialChange.kt

sealed interface FeedsPartialChange {
    fun reduce(oldState: NewsState): NewsState
}

sealed class Init : FeedsPartialChange {
    override fun reduce(oldState: NewsState): NewsState = when (this) {
        Loading -> oldState.copy(isLoading = true)
        is Success -> oldState.copy(
            data = news,
            isLoading = false,
            isLoadingMore = false,
            errorMessage = ""
        )
        is Fail -> oldState.copy(
            data = emptyList(),
            isLoading = false,
            isLoadingMore = false,
            errorMessage = error
        )
    }

    object Loading : Init()
    data class Success(val news: List<News>) : Init()
    data class Fail(val error: String) : Init()
}

sealed class More : FeedsPartialChange {
    override fun reduce(oldState: NewsState): NewsState = when (this) {
        Loading -> oldState.copy(
            isLoading = false,
            isLoadingMore = true,
            errorMessage = ""
        )
        is Success -> oldState.copy(
            data = oldState.data + news,
            isLoading = false,
            isLoadingMore = false,
            errorMessage = ""
        )
        is Fail -> oldState.copy(
            isLoadingMore = false,
            isLoading = false,
            errorMessage = error
        )
    }

    object Loading : More()
    data class Success(val news: List<News>) : More()
    data class Fail(val error: String) : More()
}

sealed class Report : FeedsPartialChange {
    override fun reduce(oldState: NewsState): NewsState = when (this) {
        is Success -> oldState.copy(
            data = oldState.data.filterNot { it.id == id },
            reportToast = "举报成功"
        )
        Fail -> oldState.copy(reportToast = "举报失败")
    }

    class Success(val id: Long) : Report()
    object Fail : Report()
}

作者：唐子玄
链接：https://juejin.cn/post/7108498411149590558