如何避免在 Spring Webflux 中两次读取正文？答案

【问题标题】：How to avoid reading body twice in Spring Webflux?如何避免在 Spring Webflux 中两次读取正文？
【发布时间】：2019-10-06 06:59:34
【问题描述】：

我在 body 中有一个对象 MultiMeasurement，它有一个 Status 和 1 到 n 测量值。

我必须将状态和测量值保存在存储库中，使用它们发出事件并从端点返回正确的响应。

fun save(multiMeasurement: Mono<MultiMeasurement>) =
    multiMeasurement.map(MultiMeasurement::validate)
        .flatMapMany { measurementCollector.save(it) }
        .doOnNext { eventSender.sendEvent(it.toMeasurementItem()) }
        .then(statusFacade.save(multiMeasurement))
        .doOnNext { eventSender.sendEvent(it.toStatusUpdateItem()) }
        .flatMap { toSettings(it.deviceId) }

以下代码的问题是正文被读取了两次 - 第一次在 map 运算符中，第二次在 then 中。

两者 - statusFacade 和 measurementCollector 都以反应方式 (mongo) 保存对象并返回适当的 Monos。

onOperatorDebug 的结果：

|_  Mono.flatMap ⇢ com.openbrewery.beerbob.status.StatusCollector.save(StatusCollector.kt:15)
|_  Flux.then ⇢ com.openbrewery.beerbob.measurement.MeasurementFacade.save(MeasurementFacade.kt:21)
|_  Mono.doOnNext ⇢ com.openbrewery.beerbob.measurement.MeasurementFacade.save(MeasurementFacade.kt:22)
|_  Mono.flatMap ⇢ com.openbrewery.beerbob.measurement.MeasurementFacade.save(MeasurementFacade.kt:23)

有什么方法可以读取一次正文并在measurementCollector 和statusFacade 中重复使用它？

【问题讨论】：

statusFacade.save 是否接受 MultiMeasurement？还是只有Mono<MultiMeasurement>？
两者都可以，因为这是响应式 mongo 存储库
multiMeasurement 的 flatMap 是否解决了您的问题？
我在所有操作之前在multiMeasurement 上添加了cache()

标签： kotlin spring-webflux project-reactor reactor-netty

【解决方案1】：

您可以尝试flatMap multiMeasurement 单声道，它会读取一次正文。在 flatMap 中，您的代码的其余部分应该非常相似。应该是这样的：

fun save(multiMeasurement: Mono<MultiMeasurement>) = multiMeasurement
    .flatMap { mm ->
        measurementCollector
            .save(mm.validate())
            .doOnNext { eventSender.sendEvent(it.toMeasurementItem()) }
            .then(statusFacade.save(mm))
            .doOnNext { eventSender.sendEvent(it.toStatusUpdateItem()) }
            .flatMap { toSettings(it.deviceId) }
    }

在您的代码中，multiMeasurement mono 被订阅了两次，这就是您阅读正文两次的原因。为避免这种情况，您需要将单声道的值平展一次，然后将实际值向前传递。

【讨论】：