使用 buffer() 处理 Kotlin 流异常

Question

我有一个在第二个元素上引发异常的简单流程：

private val log = LoggerFactory.getLogger("test")

fun main() = runBlocking {
    flowOf(1, 2, 3)
        .onCompletion { log.info("Flow completed${if (it == null) "" else " exceptionally: ${it.message}"}") }
        .buffer()
        .map { throwingMapper(it) }
        .catch { log.error("Exception thrown") }
        .collect { log.info(it) }
    }
}

fun throwingMapper(it: Int): String {
    if (it == 2) {
        throw Exception("Test exception")
    }
    return "$it-mapped"
}

当我执行此代码时，我得到以下输出 - 流程完成无异常：

2021-04-09 12:35:00.875 [main] INFO  - test:31 - Flow completed
2021-04-09 12:35:00.904 [main] INFO  - test:133 - 1-mapped
2021-04-09 12:35:00.915 [main] ERROR - test:34 - Exception thrown

但是，当我将 map 运算符移到 buffer 之前：

flowOf(1, 2, 3)
        .onCompletion { log.info("Flow completed${if (it == null) "" else " exceptionally: ${it.message}"}") }
        .map { throwingMapper(it) }
        .buffer()
        .catch { log.error("Exception thrown") }
        .collect { log.info(it) }

产生以下输出并完成流程例外情况：

2021-04-09 12:38:35.982 [main] INFO  - test:31 - Flow completed exceptionally: Test exception
2021-04-09 12:38:36.024 [main] ERROR - test:34 - Exception thrown

在第一种情况下，流程是如何异常完成的？ buffer 会默默吞下下游异常吗？还是它在内部创建了一个新流程？如果是这样，是否有一些保留原来的例外？

Answer 1

正如documentation 的这种方法所说：

buffer 运算符在执行期间为其应用的流程创建一个单独的协程。 [...] buffer 之前的代码将在单独的新协程 [...] 中与 [调用此代码的协程] 同时执行。

在第一种情况下，流程无异常完成，因为抛出异常的方法和捕获异常的方法都放在buffer的一侧，所以它们在同一个协程中执行。

协程之间使用通道将协程 P 发出的元素发送到协程 Q。

我认为不可能通过Channel 发送异常，因此您可能需要在buffer 之前和之后定义两个catch 处理程序（如果您希望在这两个部分都抛出异常）。