如何调试从未完成单声道

Question

有些方法返回mono 。这是一个典型的模式，分多个层：

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

假设doSomethingto()返回一个单字 （它向数据库写入一些内容，发送一条消息等）。如果我只是用mono.empty()替换它，那么一切都按预期工作，否则就不是这样了。更具体地说，单声道从未完成，它贯穿所有处理，但在结束时错过了终止信号。所以这些东西实际上是写在数据库中的，消息实际上是发送的，等等。

为了证明没有终止是问题所在，这里有一个有效的黑客：

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

null

我不能在应用程序之外用简单的单工作流复制这种行为。

Answer 1

通过在反应流中使用log()运算符，可以跟踪和记录流中的事件。这对于更好地了解应用程序中发生的事件非常有用。

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

它链接在一个序列中，窥视它上游的流量或单声道的每个事件（包括onNext、onError和onComplete以及订阅、取消和请求）。

Reactor参考文献。记录a序列