如何调试永远不会完成的 Mono答案

【问题标题】：How do I debug a Mono that never completes如何调试永远不会完成的 Mono
【发布时间】：2023-04-08 20:24:01
【问题描述】：

我有一个 Spring Boot 应用程序，其中包含一个复杂的反应流（它涉及 MongoDB 和 RabbitMQ 操作）。大多数时候它都有效，但是......

一些方法返回Mono<Void>。这是一个典型的模式，有多个层次：

fun workflowStep(things: List<Thing>): Mono<Void> =
    Flux.fromIterable(things).flatMap { thing -> doSomethingTo(thing) }.collectList().then()

假设doSomethingTo() 返回一个Mono<Void>（它将某些内容写入数据库，发送消息等）。如果我只是用Mono.empty() 替换它，那么一切都会按预期工作，否则不会。更具体地说， Mono 永远不会完成，它会运行所有处理，但最后会错过终止信号。所以事情实际上是写在数据库中，消息是实际发送的等等。

为了证明缺少终止是问题所在，这里有一个可行的技巧：

val hackedDelayedMono = Mono.empty<Void>().delayElement(Duration.ofSeconds(1))
return Mono.first(
    workflowStep(things),
    hackedDelayedMono
)

问题是，我可以用从未完成的 Mono 做些什么来弄清楚发生了什么？我无处可以放置日志语句或刹车点，因为：

没有错误
没有信号发出

如何检查 Mono 正在等待完成的内容？

ps。我无法使用简单的 Mono 工作流程在应用程序之外重现此行为。

【问题讨论】：

从函数返回Mono#empty 将具有返回类型Mono<Void>，这意味着您需要向我们展示doSomething 中的内容。您很可能在某处断开了链条，但您尚未发布可重复的工作示例，因此无法判断您在哪里断开链条。
“hackedDelayedMono”上的延迟实际上并没有做任何事情，因为延迟只发生在元素上而不是信号上。由于它是一个空单声道，它将直接在订阅时完成。

标签： spring-boot spring-webflux project-reactor

【解决方案1】：

您可以在反应流中使用log() 运算符跟踪和记录流中的事件。这有助于更好地了解您的应用中正在发生的事件。

Flux.fromIterable(things)
    .flatMap(thing -> doSomethingTo(thing))
    .log()
    .collectList()
    .then()

链接在一个序列中，它可以查看 Flux 或 Mono 的每个事件它的上游（包括 onNext、onError 和 onComplete 以及订阅、取消和请求）。

Reactor Reference Documentation - Logging a Sequence

Reactor 参考文档还包含调试反应流的其他有用建议，可在此处找到：Debugging Reactor

【讨论】：

【解决方案2】：

（我们设法解决了这个问题 - 它不是直接在我正在处理的代码中，但由于某种原因我的更改触发了它。我仍然不明白根本原因，但在链条上更高我们发现Mono.zip() 压缩Mono<Void>。虽然这以前可以工作，但它在某个时候停止工作。为什么Mono<Void> 甚至可以压缩，为什么我们没有得到编译器错误，更糟糕的是，为什么它有时会起作用吗？）

在这里回答我自己的问题，用于调试的工具将以下内容添加到链中的所有 Monos 中，直到它没有产生任何输出：

mono.doOnEach { x ->
    logger.info("signal: ${x}")
}
.then(Mono.defer {
    logger.info("then()")
    Mono.empty<Void>()
})

我还尝试了.log() - 也是很好的工具，但可能太详细了，而且很难理解哪个 Mono 产生了哪些日志消息 - 因为这些是使用动态范围而不是词法范围记录的，上面的方法清楚地告诉你。

【讨论】：