kotlin - 等待函数(协程)和订阅 Reactor 有什么区别

Question

我正在使用 Kotlin + Reactor(Mono 和 Flux)，我想知道使用 await...() 之间的区别(来自 kotlin-coroutines-reactive )函数和 subscribe() (来自 react 堆)。我举了两个例子来展示我正在尝试做的事情。

示例 1(带有等待功能):

@Test
internal fun test() = runBlockingTest {
    Mono.error<String>(IllegalStateException("exception"))
        .doOnError {
            print("error")
        }.awaitFirst().let {
            print("success")
        }
}

输出:“错误” 带有 IllegalStateException 堆栈跟踪 .

示例 2(带有订阅功能):

@Test
internal fun test() = runBlockingTest {
    Mono.error<String>(IllegalStateException("exception"))
        .doOnError {
            print("error")
        }.subscribe {
            print("success")
        }
}

输出:
只是“错误”。

为什么示例 1 显示堆栈跟踪而示例 2 不显示？

谢谢。

Answer 1

当您调用 subscribe在 react 链上，您将其与主流解耦，它变得独立并且可能是异步的。错误作为信号而不是传统上抛出的异常在 react 流上传播。在这种情况下，错误处理是使用 doOnError 等运算符的 react 链的责任。 , onErrorMap , onErrorReturn ， ETC。

另一方面，Kotlin 的 await打破这种独立性并将 react 流附加回主流，并允许您编写 react /异步代码，因为它是命令式的(例如:try-catch block ，解包函数返回类型等)。