Spring Webflux 正确的查找和保存方式

Question

我创建了以下方法来查找 Analysis 对象，更新其上的结果字段，然后最后将结果保存在数据库中但不等待 return。

public void updateAnalysisWithResults(String uuidString, String results) {
        findByUUID(uuidString).subscribe(analysis -> {
            analysis.setResults(results);
            computeSCARepository.save(analysis).subscribe();
        });
    }

在订阅中订阅感觉写得不好。这是一种不好的做法吗？有没有更好的写法？

更新：入口点

@PatchMapping("compute/{uuid}/results")
    public Mono<Void> patchAnalysisWithResults(@PathVariable String uuid, @RequestBody String results) {
        return computeSCAService.updateAnalysisWithResults(uuid,results);
    }

    public Mono<Void> updateAnalysisWithResults(String uuidString, String results) {
//        findByUUID(uuidString).subscribe(analysis -> {
//            analysis.setResults(results);
//            computeSCARepository.save(analysis).subscribe();
//        });
        return findByUUID(uuidString)
                .doOnNext(analysis -> analysis.setResults(results))
                .doOnNext(computeSCARepository::save)
                .then();
    }

Answer 1

在订阅中订阅不是一个好习惯。您可以使用 flatMap 运算符来解决此问题。

public void updateAnalysisWithResults(String uuidString, String results) {
        findByUUID(uuidString).flatMap(analysis -> {
            analysis.setResults(results);
            return computeSCARepository.save(analysis);
        }).subscribe();
    }

Answer 2

为什么它不起作用是因为你误解了 doOnNext 的作用。

让我们从头开始。

一个Flux或Mono是生产者，他们生产物品。您的应用程序向调用客户端生成内容，因此它应该始终 return Mono 或 Flux。如果你不想 return 任何你应该 return a Mono<Void>.

当客户端 subscribes 访问您的应用程序时，reactor 将做的是沿相反方向调用所有运算符，直到找到 producer。这就是所谓的assembly phase。如果你所有的运营商不链接在一起，你就是我所说的 breaking the reactive chain.

断链后，断链的东西不会执行

如果我们看一下您的示例，但在更详细的版本中：

@Test
void brokenChainTest() {
    updateAnalysisWithResults("12345", "Foo").subscribe();
}

public Mono<Void> updateAnalysisWithResults(String uuidString, String results) {
    return findByUUID(uuidString)
            .doOnNext(analysis -> analysis.setValue(results))
            .doOnNext(this::save)
            .then();
}

private Mono<Data> save(Data data) {
    return Mono.fromCallable(() -> {
        System.out.println("Will not print");
        return data;
    });
}

private Mono<Data> findByUUID(String uuidString) {
    return Mono.just(new Data());
}

private static class Data {
    private String value;

    public void setValue(String value) {
        this.value = value;
    }
}

在上面的例子中 save 是一个 callable 函数，它会 return 一个 producer。但是如果我们运行上面的函数你会注意到print永远不会被执行。

这与doOnNext的用法有关。如果我们阅读它的文档，它会说：

Add behavior triggered when the Mono emits a data successfully. The Consumer is executed first, then the onNext signal is propagated downstream.

doOnNext 接受 Consumer 而 return 无效。如果我们查看 doOnNext，我们会看到函数描述如下所示：

public final Mono<T> doOnNext(Consumer<? super T> onNext)`

这意味着它接受一个 T 或扩展 T 的消费者，它 return 是一个 Mono<T>。因此，为了保持长的解释简短，您可以看到它消耗了一些东西，但也 return 是相同的东西。

意思是说这通常用于所谓的side effects，基本上是指边不妨碍电流的事情。其中之一可以是日志记录。日志记录是那些会消耗例如字符串并记录它的东西之一，而我们希望让字符串在我们的程序中流动。或者我们可能想在侧面增加一个数字。或者在某处修改一些状态。您可以阅读有关副作用的所有信息 here.

你可以这样直观地想它：

     _____ side effect (for instance logging)
    /
___/______ main reactive flow

这就是为什么您的第一个 doOnNext setter 有效，因为您 一边修改状态 ，一边设置 class 因此将 class 的状态修改为具有值。

另一方面，第二条语句 save 没有被执行。你看那个函数实际上是 returning 我们需要处理的事情。

这是它的样子：

      save
     _____ 
    /     \   < Broken return
___/             ____ no main reactive flow

我们所要做的实际上是更改一行：

// From
.doOnNext(this::save)

// To
.flatMap(this::save)

flatMap 获取 Mono 中的任何内容，然后我们可以使用它来执行某些操作，然后 return 一个“新的”操作。

所以我们的流程（使用 flatMap）现在看起来像这样：

   setValue()    save()
    ______       _____ 
   /            /     \   
__/____________/       \______ return to client

因此，通过使用 flatMap，我们现在正在保存并 returning 从该函数触发链的其余部分 returned 的任何内容。

如果您随后选择忽略 return 从 flatMap 中编辑的任何内容，则完全正确，就像您调用 then 所做的那样，这将

Return a Mono which only replays complete and error signals from this

一般规则是，在完全响应式应用程序中，你永远不应该阻塞。

而且您通常不会 subscribe 除非您的申请是最终申请 consumer。这意味着如果您的应用程序启动了请求，那么您就是其他东西的 consumer，所以您 subscribe。如果网页从请求开始，那么他们就是最终的 consumer 并且他们正在订阅。

如果您在正在生成数据的应用程序中订阅，就像您在运行开一家面包店并同时吃烤面包一样。

不要那样做，这对生意不利:D

Spring Webflux 正确的查找和保存方式

Spring Webflux Proper Way To Find and Save

java

reactive-streams

spring-webflux