Java反应流在流的下一步还需要映射对象时如何映射对象

Question

我正在使用Java11和ProjectReactor（来自Spring）。我需要对RESTAPI进行http调用（在整个流中只能进行一次）。对于响应，我需要计算两件事：

1. 检查数据库（mongodb）中是否存在文档。如果它不存在，则创建并返回它。否则就退回吧

在伪代码中是这样的：

public void computeData(String id) {
    httpClient.getData(id) // Returns a Mono<Data>
        .flatMap(data -> getDocument(data.getDocumenId()))
         // Issue here is we need access to the data object consumed in the previous flatMap but at the same time we also need the document object we get from the previous flatMap
        .flatMap(document -> calculateValue(document, data)) 
        .subscribe();
}

public Mono<Document> getDocument(String id) {
    // Check if document exists
    // If not create document

    return document;
}

public Mono<Value> calculateValue(Document doc, Data data) {
    // Do something...
    return value;
}

问题是calculateValue需要来自http的返回值。getData，但这已经在第一个flatMap上使用，但我们还需要从上一个flatMap获取的文档对象。

我试图用Mono解决这个问题。邮政编码如下：

public void computeData(String id) {
    final Mono<Data> dataMono = httpClient.getData(id);

    Mono.zip(
        new Mono<Mono<Document>>() {
            @Override
            public void subscribe(CoreSubscriber<? super Mono<Document>> actual) {
                final Mono<Document> documentMono = dataMono.flatMap(data -> getDocument(data.getDocumentId()))
                actual.onNext(documentMono);
            }
        },
        new Mono<Mono<Value>>() {
            @Override
            public void subscribe(CoreSubscriber<? super Mono<Value>> actual) {
                actual.onNext(dataMono);
            }
        }
    )
    .flatMap(objects -> {
        final Mono<Document> documentMono = objects.getT1();
        final Mono<Data> dataMono = objects.getT2();

        return Mono.zip(documentMono, dataMono, (document, data) -> calculateValue(document, data))
    })
}

但这是在执行httpClient。getData（id）两次，这违反了我只调用一次的限制。我理解为什么要执行两次（我订阅了两次）。

也许我的解决方案设计可以在某个地方改进，但我看不出在哪里。对我来说，在设计反应式代码时，这听起来像是一个“正常”问题，但到目前为止，我还没有找到合适的解决方案。

我的问题是，如何以反应式和非阻塞的方式完成这个流程，并且只对RESTAPI进行一次调用？

附言我可以在一个映射中添加所有逻辑，但这会迫使我订阅映射中的一个Mono，这是不推荐的，我希望避免采用这种方法。

编辑关于@caco3注释我需要在地图内部订阅，因为getDocument和calculateValue方法都返回Mono。

所以，如果我想把所有的逻辑放在一张地图里，它会是这样的：

public void computeData(String id) {
    httpClient.getData(id)
        .map(data -> getDocument(data).subscribe(s -> calculateValue(s, data)))
        .subscribe();
}




Answer 1

简单地说，你的问题类似于：

Mono.just(1)
        .flatMap(original -> process(original))
        .flatMap(processed -> I need access to the original value and the processed value!
                System.out.println(original); //Won't work
        );


private static Mono<String> process(int in) {
    return Mono.just(in + " is an integer").delayElement(Duration.ofSeconds(2));
}

（愚蠢的例子，我知道。）

问题是map（）（扩展来说，platMap（））是转换-您可以访问新值，而旧值会消失。因此，在您的第二次调用中，您可以访问1是整数，但不是原始值（1）。）

这里的解决方案是，不是映射到新值，而是映射到某种包含原始值和新值的合并结果。反应器为此提供了一个内置类型-一个Tuple。所以编辑我们最初的例子，我们需要：

Mono.just(1)
        .flatMap(original -> operation(original))
        .flatMap(processed -> //Help - I need access to the original value and the processed value!
                System.out.println(processed.getT1()); //Original
                System.out.println(processed.getT2()); //Processed

                ///etc.
        );


private static Mono<Tuple2<Integer, String>> operation(int in) {
    return Mono.just(in + " is an integer").delayElement(Duration.ofSeconds(2))
            .map(newValue -> Tuples.of(in, newValue));
}

您可以使用相同的策略来保持两个文档和数据-不需要内部订阅或任何排序：-）

Answer 2

您不必在map中订阅，只需继续在flatMap中构建反应链即可：

getData(id) // Mono<Data>
        .flatMap(data -> getDocument(data.getDocumentId()) // Mono<Document>
                .switchIfEmpty(createDocument(data.getDocumentId())) // Mono<Document>
                .flatMap(document -> calculateValue(document, data)) // Mono<Value>
         )
        .subscribe()