执行多个反应操作的正确方法答案

【问题标题】：Correct way to execute multiple reactive operations执行多个反应操作的正确方法
【发布时间】：2019-04-20 01:42:51
【问题描述】：

我从响应式存储库中获得Mono<FooBar>，基于它的值我必须创建另外两个对象使用响应式存储库保存它们，修改FooBar 对象并保存它。

由于我是响应式编程的新手，因此我使用了以下解决方案，该解决方案有效，但我不确定我是否正确使用了响应式 API：

@Test
void createAndSave() {
    Mono<FooBar> fooBarMono = findFooBar()  // returns Mono<FooBar>
            .map(fooBar -> {
        createAndSaveLoremBar(fooBar).subscribe();   // returns Mono<LoremBar>
        createAndSaveDoloremBar(fooBar).subscribe(); // returns Mono<DoloremBar>

        fooBar.setActive(true);

        return saveFooBar(fooBar);          // returns Mono<FooBar>
    }).flatMap(Function.identity());

    StepVerifier.create(fooBarMono)
            .expectNextMatches(Objects::nonNull)
            .expectComplete()
            .verify();

}

来自控制台日志：

   saved lorem bar
   saved dolorem bar
   saved foo bar

【问题讨论】：

激活fooBar前是否需要保存lorem bar和dolorem bar？那么 lorem 和 dolorem 保存的错误呢？我们应该忽略还是传播错误？
@AlexanderPankin 保存顺序无关紧要；如果可能，应该传播错误

标签： spring-data-mongodb spring-webflow project-reactor

【解决方案1】：

我认为下面的解决方案更具可读性。无论如何亚历山大是正确的，你永远不应该修改输入。你看我们从函数式编程中借用了很多概念。例如，您调用 Function.identity() 这被称为 identity functor。 Flux 和 Mono 都是单子。还有一个对这些人来说是秘密的概念，称为Referential transparency，即命令式更新中断。

    final Mono<FooBar> fooBarMono1 = findFooBar()
            .zipWhen((fooBar) -> createAndSaveLoremBar(fooBar))
            .map(tuple -> tuple.getT1())
            .zipWhen((fooBar) -> createAndSaveDoloremBar(fooBar))
            .map(tuple -> tuple.getT1())
            .map(fooBar -> new FooBar(true))
            .flatMap(fooBar -> saveFooBar(fooBar));

或者更简洁：

    final Mono<FooBar> fooBarMono1 = findFooBar()
            .zipWhen((fooBar) -> createAndSaveLoremBar(fooBar)
                                    .then(createAndSaveDoloremBar(fooBar)))
            .map(tuple -> tuple.getT1())
            .map(fooBar -> new FooBar(true))
            .flatMap(fooBar -> saveFooBar(fooBar));

【讨论】：

【解决方案2】：

起初，在异步（反应式）世界中改变对象并不是一个好主意。

无论如何，在您的解决方案中，lorem 和 dolorem 保存可能出现的错误将被忽略。你可以像这样改进它：

Mono<FooBar> fooBarMono = findFooBar()
        .flatMap(fooBar -> Flux.merge(
                createAndSaveLoremBar(fooBar),
                createAndSaveDoloremBar(fooBar)) // asynchronously saving lorem and dolorem
                .then(Mono.fromCallable(() -> {  // if there wasn't errors, mutate and save fooBar
                    fooBar.setActive(true);
                    return fooBar;
                }).flatMap(fooBar1 -> saveFooBar(fooBar1))));

如果您可以使用真正的active 标志创建fooBar 的副本，则代码可能会更简单。例如龙目岛。

@Builder(toBuilder = true)
public class FooBar {
...
}

Mono<FooBar> fooBarMono = findFooBar()
        .flatMap(fooBar -> Flux.merge(
                createAndSaveLoremBar(fooBar),
                createAndSaveDoloremBar(fooBar))
                .then(saveFooBar(fooBar.toBuilder().active(true).build())));

如果您对 saveFooBar(...) 的结果不感兴趣，而只对完成信号感兴趣，则可以异步进行所有三个保存：

Flux<Object> flux = findFooBar()
        .flatMapMany(fooBar -> Flux.merge(
                createAndSaveLoremBar(fooBar),
                createAndSaveDoloremBar(fooBar),
                saveFooBar(fooBar.toBuilder().active(true).build())));

实际上，在最后一种方法中，您可以收集所有三个结果，并且您应该更喜欢这种方法，但是我没有足够的关于您的类和创建完整示例的要求的信息。

【讨论】：