如何在 Reactor 的上下文中保存返回的方法 mono，以便下次在此 Reactor 流中调用该方法时使用它？

Question

我有一个从一些repo返回数据的方法。当然，这些数据可以随时更改。我希望确保从该方法返回的数据在操作链期间保持一致(该方法也可以在操作链中从另一个组件调用(。例如：

fetchDataThanCanBeChanged()
.flatMap(...)
.doOnNext(...)
.doOnSuccess(...)
.map(...)
.zipWith(...)
.flatMap(...)
.then(fetchDataThanCanBeChanged())......

我想确保对fetchDataThanCanBeChanged((的第二次调用返回的数据与第一次相同。我试图用Mono.ferContextual:方法包装fetchDataThatCanBeChanged

private Mono<Object> fetchDataUsingCache() {
return fetchDataThatCanBeChanged()
.flatMap(s ->
Mono.deferContextual(ctx -> {
System.out.println("cached snapshot=" + ctx.getOrEmpty("KEY")); 
return Mono.just(s);
}).contextWrite(ctx ->
ctx.put("KEY", ctx.getOrDefault("KEY", s))
));
}

并称之为两次：

fetchDataUsingCache()
.flatMap(d -> change(d))
.doOnNext(p -> saveChangedDataInRepo(p))
.delayElement(Duration.ofMillis(2000))
.fetchDataUsingCache()
.block();

但是fetchDataThatCanBeChanged((的内容只执行一次，返回的数据是更新的。

有解决方案的想法吗？？？非常感谢！

Answer 1

最后，我用Mono.deferContextual将我的代码包装在需要旧的提取数据的地方，并将数据写入如下上下文：

fetchDataThanCanBeChanged()
.flatMap(s -> Mono.deferContextual(ctx -> 
changeDataInRepo()
.doOnSuccess(b -> fetchDataThanCanBeChanged())
.thenReturn(doWhateverNeedWithOldData(ctx.getOrEmpty("KEY"))))
.contextWrite(ctx -> ctx.put("KEY", s))))
.block();