带有catch操作符的Kotlin流仍然完成

Question

我很难理解catch算子在kotlinFlow中的工作方式。下面是catch文档

问题:

1. 为什么catch的存在不允许Flow在遇到异常时继续，而不是完成?
2. catch操作符的位置似乎改变了行为。为什么我不能把catch操作符在链的末尾看到相同的结果?在我的例子中，只有当我把它放在onEach之前，它才会执行。

例子要点

第一个例子，将catch放在onEach之前:

fun main() {
// Flow of lambdas that return a String (or throw an Exception)
flowOf<() -> String>({ "Hello " }, { error("error") }, { "World" })
// Map to the result of the invocation of the lambda
.map { it() }
// This line will emit the error String, but then the flow completes anyway.
// I would expect the flow to continue onto "World"
.catch { emit("[Exception caught] ") }
.onEach { println(it) }
.launchIn(GlobalScope)
}

实际结果:Hello [Exception caught]

预期结果:Hello [Exception caught] World

第二个例子，将catch放在onEach之后:

fun main() {
// Flow of lambdas that return a String (or throw an Exception)
flowOf<() -> String>({ "Hello " }, { error("error") }, { "World" })
// Map to the result of the invocation of the lambda
.map { it() }
.onEach { println(it) }
// I would expect this catch to emit, but it never gets here.
.catch { emit("[Exception caught] ") }
.launchIn(GlobalScope)
}

实际结果:Hello

预期结果:Hello [Exception caught] World

或者，由于onEach发生在catch发射之前，因此catch的发射将被忽略?在哪种情况下，预期输出是这样的?：Hello World

Answer 1

对我来说，解释你在做什么最简单的方法就是把它简化成同步代码。你基本上是这样做的:

fun main() {
val list = listOf("Hello", "error", "World")

try {
for (s in list) {
if (s == "error") error("this is the error message here")
println(s)
}
} catch (e: Exception) {
println("the exception message is: ${e.localizedMessage}")
}
}

输出:

Hello
the exception message is: this is the error message here

可以看到，异常被捕获了，但是它不能阻止for循环的停止。与异常停止map函数的方式相同。

流。Catch将捕获一个异常并阻止它的传播(除非你再次抛出它)，但它不能后退一步(到map fun)，并告诉它从下一个元素的位置神奇地重新开始，等等。

如果你想这样做，你需要在.mapfun里面放一个正常的try/catch。所以应该是:

fun main() {
val list = listOf("Hello", "error", "World")

for (s in list) {
try {
if (s == "error") error("this is the error message here")
println(s)
} catch (e: Exception) {
println("the exception message is: ${e.localizedMessage}")
}
}
}

输出:

Hello
the exception message is: this is the error message here
World

通常使用Flow.catch的方式是捕获一个异常，以阻止下一步，就像如果你有:

//pseudo
flow
.map{/*code*/}
.filterNotNull()
.doSomethingRisky() //this can throw an exception
.catch {} //do something about it
.doSomethingElse()

在这种情况下，即使doSomethingRisky抛出异常，流仍然会到达doSomethingElse。这或多或少就是Flow.catch的用法。

Answer 2

虽然AlexT响应是完全正确的，但有一个技巧可以保持流的活动。您可以将实际的流包装到另一个流中。内部/实际流将完成，但外部流将保持活跃。如果你的流程失败，但你想提供"重试"，这很方便。按钮:

private val sourceFlow = MutableStateFlow(0)
private val innerFlow = flow { ... } // Whatever your flow is!
val outerFlow = sourceFlow.flatMapLatest {
innerFlow.catch {
// Handling, e.g. emitting an error state to the UI (can show retry button)
emit(ViewState.Error("Some error")
}.onStart {
println("Inner start")
}.onCompletion {
println("Inner end")
}
}.onStart {
println("Outer start")
}.onCompletion {
println("Outer end")
}

如果您想重试整个流程，只需执行sourceFlow.update { it + 1 }。这将更新MutableStateFlow中的0，导致再次调用flatMapLatest，然后再次启动innerFlow。您可以使用MutableSharedFlow而不是MutableStateFlow<Int>执行相同的操作，但我发现这有点棘手，因为在这种情况下，您需要从协程发出。

使用此设置，您需要订阅outerFlow，它将发出innerFlow的所有值