Kotlin Flow处理错误后为何不再触发flatMapLatest转换函数？

原始代码 #

ViewModel 代码

class MarketViewModel @Inject constructor(repo: MarketRepository) : ViewModel() {
    private val retry = MutableStateFlow(0)

    val marketState: LiveData<State<Market>> =
        retry.flatMapLatest{repo.refreshMarket()}
        .map { State.Success(it) as State<T> }
        .catch { error -> emit(State.Error(error)) }
        .stateIn(vmScope, SharingStarted.WhileSubscribed(5000), State.Loading())
        .asLiveData()

    fun retry() {
        retry.value++
    }
}

Repository 代码

fun refreshMarket() =
    flow { emit(api.getMarkets()) }
        .onEach { db.upsert(it) }
        .flowOn(dispatchers.IO)

问题原因分析 #

1. 为什么调用retry()后flatMapLatest不再触发？
   当前代码中，catch操作符位于flatMapLatest和map之后，属于外层流的异常处理逻辑。当refreshMarket抛出网络异常时，catch会捕获异常并发射State.Error，但之后整个外层流（由retry StateFlow驱动的完整流链）会被直接终止。这导致retry后续的value变化（调用retry()时的自增操作）无法再触发flatMapLatest，因为外层流已经处于完成状态。

2. catch块会导致流完成吗？
   是的。默认情况下，catch捕获异常后会终止当前所在的流。如果catch作用于外层流，整个流会停止发射新值；只有当catch作用于内层流时，才只会终止该内层流，外层流仍能继续响应新的触发信号。

修复方案 #

将catch操作符移到flatMapLatest内部，针对每个refreshMarket返回的流做独立异常处理，确保单个内层流的异常不会影响外层的retry StateFlow。修改后的ViewModel代码如下：

class MarketViewModel @Inject constructor(repo: MarketRepository) : ViewModel() {
    private val retry = MutableStateFlow(0)

    val marketState: LiveData<State<Market>> =
        retry.flatMapLatest { 
            repo.refreshMarket()
                .map { State.Success(it) as State<Market> }
                .catch { error -> emit(State.Error(error)) }
        }
        .stateIn(vmScope, SharingStarted.WhileSubscribed(5000), State.Loading())
        .asLiveData()

    fun retry() {
        retry.value++
    }
}

修复说明 #

• 现在每个由refreshMarket生成的流都自带异常处理，当某个流抛出异常时，catch只会处理该流的异常并发射State.Error，随后该内层流结束，但外层的retry StateFlow仍然保持活跃状态。
• 调用retry()时，retry.value自增会触发flatMapLatest重新执行repo.refreshMarket()，发起新的网络请求，正常实现重试功能。

内容的提问来源于stack exchange，提问作者Samuel Robert