简单的 for 循环 java reactor post flatmap答案

【问题标题】：simple for loop java reactor post flatmap简单的 for 循环 java reactor post flatmap
【发布时间】：2023-01-12 18:09:16
【问题描述】：

我的问题是我做了一个 post 请求来获取我的数据库中的元素总数，我需要做一个 for 循环直到我达到那个整数除法 10。

我当前不工作的代码

protected Mono<List<Long>> getAllSubscriptionIds(ProductCode productCode) {
    List<Long> subscriptionIds = new ArrayList<>();

    String body = "{\n" +
            " \"productCodes\": [\"" + productCode.name() + "\"],\n" +
            " \"pagination\": {\n" +
            "     \"offset\": 0,\n" +
            "     \"limit\": 10" +
            "\n  }\n" +
            " }";
    //first post where I get the number of elements in my db
    return restClient.post(
                    "https://" + url,
                    buildRequiredHeaders(),
                    body,
                    String.class
            )
            .onErrorMap(err-> new RuntimeException(err.getMessage()))
            .flatMap(response -> {
                log.debug(response);
                ResponseModel<DataLakeCallResponse<JsonNode>> variable = null;
                try {
                    variable = JsonUtil.fromString(response, new TypeReference<ResponseModel<DataLakeCallResponse<JsonNode>>>() {
                    });
                    log.debug(response);
                } catch (IOException e) {
                    throw new RuntimeException(e);
                }
                variable.getPayload().getList().forEach(
                        object-> subscriptionIds.add(object.get("subscriptionId").asLong()));
                //if number of elements > 10
                if(variable.getPayload().getPagination().getResultCount() > 10){
                    //for loop on that number / 10 (so that I can use an offset
                    for (int i = 0; i < variable.getPayload().getPagination().getResultCount() / 10; i++){
                        String bodyI = "{\n" +
                                " \"productCodes\": [\"" + productCode.name() + "\"],\n" +
                                " \"pagination\": {\n" +
                                "     \"offset\": " + (i + 1) * 10 + ",\n" +
                                "     \"limit\": 10\n" +
                                "  }\n" +
                                " }";
                        return restClient.post(
                                        "https://" + url,
                                        buildRequiredHeaders(),
                                        bodyI,
                                        String.class
                                )
                                .onErrorMap(err-> new RuntimeException(err.getMessage()))
                                .flatMap(resp -> {
                                    ResponseModel<DataLakeCallResponse<JsonNode>> varia = null;
                                    try {
                                        varia = JsonUtil.fromString(resp, new TypeReference<ResponseModel<DataLakeCallResponse<JsonNode>>>() {
                                        });
                                    } catch (IOException e) {
                                        throw new RuntimeException(e);
                                    }
                                    varia.getPayload().getList().forEach(
                                            object-> subscriptionIds.add(object.get("subscriptionId").asLong()));

                                    return Mono.just(subscriptionIds);
                                });
                    }
                }
                return Mono.just(subscriptionIds);
            });
}

我确实理解为什么这不起作用（它在 for 循环内返回）但我真的不明白我可以使用什么替代方法来使其起作用。我尝试了一种外部方法，但它仍然会失败。我尝试了 Mono.zip，但我想我试错了。

这是我尝试过的替代方法，但仍然无效。

protected Mono<Object> getAllSubscriptionIds(ProductCode productCode) {
this.counter = 0;
List<Long> subscriptionIds = new ArrayList<>();
List<Mono<Integer>> resultList = new ArrayList<>();

String body = "{\n" +
        " \"productCodes\": [\"" + productCode.name() + "\"],\n" +
        " \"pagination\": {\n" +
        "     \"offset\": 0,\n" +
        "     \"limit\": 10" +
        "\n  }\n" +
        " }";

return restClient.post(
                "https://" + url,
                buildRequiredHeaders(),
                body,
                String.class
        )
        .onErrorMap(err-> new RuntimeException(err.getMessage()))
        .flatMap(response -> {
            log.debug(response);
            ResponseModel<DataLakeCallResponse<JsonNode>> variable = null;
            try {
                variable = JsonUtil.fromString(response, new TypeReference<ResponseModel<DataLakeCallResponse<JsonNode>>>() {
                });
                log.debug(response);
            } catch (IOException e) {
                throw new RuntimeException(e);
            }
            variable.getPayload().getList().forEach(
                    object-> subscriptionIds.add(object.get("subscriptionId").asLong()));

            if(variable.getPayload().getPagination().getResultCount() > 10){
                for (int i = 0; i < variable.getPayload().getPagination().getResultCount() / 10; i++){
                    resultList.add(Mono.just(i));
                }
            }

            return Mono.zip(resultList, intMono -> {
                this.counter++;
                String bodyI = "{\n" +
                        " \"productCodes\": [\"" + productCode.name() + "\"],\n" +
                        " \"pagination\": {\n" +
                        "     \"offset\": " + this.counter * 10 + ",\n" +
                        "     \"limit\": 10\n" +
                        "  }\n" +
                        " }";
                return restClient.post(
                                "https://" + url,
                                buildRequiredHeaders(),
                                bodyI,
                                String.class
                        )
                        .onErrorMap(err-> new RuntimeException(err.getMessage()))
                        .flatMap(resp -> {
                            ResponseModel<DataLakeCallResponse<JsonNode>> varia = null;
                            try {
                                varia = JsonUtil.fromString(resp, new TypeReference<ResponseModel<DataLakeCallResponse<JsonNode>>>() {
                                });
                            } catch (IOException e) {
                                throw new RuntimeException(e);
                            }
                            varia.getPayload().getList().forEach(
                                    object-> subscriptionIds.add(object.get("subscriptionId").asLong()));

                            return Mono.just(subscriptionIds);
                        });
            });
           // return Mono.just(subscriptionIds);
        });
}

知道如何解决这个问题吗？

【问题讨论】：

标签： java reactor flatmap

【解决方案1】：

您的代码的问题是您在 for 循环内返回，这导致函数在循环的第一次迭代后立即返回。您可以使用 flatMap 运算符来保持管道运行并将每次迭代的结果添加到 subscriptionIds 而不是返回

【讨论】：

问题是平面图需要返回，我不知道如何避免。你能分享你的解决方案来避免它吗？
我还在第一个 flatMap 之后尝试了 .flatMap 但我不知道如何重复这种行为我多次（我在第一篇文章的第一个循环结果中使用的整数值）
好的，我明白了，你也可以使用 flatMapMany() 和 Flux.range() 而不是 for 循环，你需要用 `Flux.range(1, totalPages).flatMapMany(i ->` 替换 for 循环，你也需要在 flatMapMany 之后返回 `Mono.just(subscriptionIds)`
谢谢，明天早上我会尝试这个解决方案（这里是晚上）。
在 Flux.range 之后不能使用 flatMapMany，它只接受 flatMap。

【解决方案2】：

好的，我终于找到了解决方案

protected Flux<Object> getAllSubscriptionIds(ProductCode productCode) {
    List<Long> subscriptionIds = new ArrayList<>();
    AtomicInteger i = new AtomicInteger();

    String body = "{
" +
            " "productCodes": ["" + productCode.name() + ""],
" +
            " "pagination": {
" +
            "     "offset": 0,
" +
            "     "limit": 1000" +
            "
  }
" +
            " }";

    return restClient.post(
                    "https://" + url,
                    buildRequiredHeaders(),
                    body,
                    String.class
            )
            .onErrorMap(err-> new RuntimeException(err.getMessage()))
            .flatMapMany(response -> {
                log.debug(response);
                ResponseModel<DataLakeCallResponse<JsonNode>> variable = null;
                try {
                    variable = JsonUtil.fromString(response, new TypeReference<ResponseModel<DataLakeCallResponse<JsonNode>>>() {
                    });
                    log.debug(response);
                } catch (IOException e) {
                    throw new RuntimeException(e);
                }
                variable.getPayload().getList().forEach(
                        object-> subscriptionIds.add(object.get("subscriptionId").asLong()));

                if(variable.getPayload().getPagination().getResultCount() > 1000){
                        String bodyI = "{
" +
                                " "productCodes": ["" + productCode.name() + ""],
" +
                                " "pagination": {
" +
                                "     "offset": " + i.incrementAndGet() * 1000 + ",
" +
                                "     "limit": 1000
" +
                                "  }
" +
                                " }";
                        return restClient.post(
                                        "https://" + url,
                                        buildRequiredHeaders(),
                                        bodyI,
                                        String.class
                                )
                                .onErrorMap(err-> new RuntimeException(err.getMessage()))
                                .flatMap(resp -> {
                                    return restClient.post(
                                                    "https://" + url,
                                                    buildRequiredHeaders(),
                                                    "{
" +
                                                            " "productCodes": ["" + productCode.name() + ""],
" +
                                                            " "pagination": {
" +
                                                            "     "offset": " + i.incrementAndGet() * 1000 + ",
" +
                                                            "     "limit": 1000
" +
                                                            "  }
" +
                                                            " }",
                                                    String.class
                                            )
                                            .onErrorMap(err-> new RuntimeException(err.getMessage()))
                                            .flatMap(respI -> {
                                                ResponseModel<DataLakeCallResponse<JsonNode>> varia = null;
                                                try {
                                                    varia = JsonUtil.fromString(respI, new TypeReference<ResponseModel<DataLakeCallResponse<JsonNode>>>() {
                                                    });
                                                } catch (IOException e) {
                                                    throw new RuntimeException(e);
                                                }
                                                varia.getPayload().getList().forEach(
                                                        object-> subscriptionIds.add(object.get("subscriptionId").asLong()));

                                                return Mono.just(subscriptionIds);
                                            });
                                }).repeat(variable.getPayload().getPagination().getResultCount() / 1000);
                }
                return Mono.just(subscriptionIds);
            });
}

基本上，我将第一个 flatMap 更改为 flatMapMany，这样我就可以在其中拥有一个带有重复循环的 flatMap。我不得不返回一个 Flux 而不是我原来的 Mono<List> 但因为我知道它总是会导致一个 Mono<List> 无论如何我将原始调用者更改为

return getAllSubscriptionIds(request.getEventMetadata().getProductCode()).collect(Collectors.reducing((i1, i2) -> i1)).flatMap(responseIds -> {
        List<BillableApiCall> queryResults = dataLakeMapper.getBillableCallsApiCheckIban(
                ((ArrayList<Long>)responseIds.get()),
                DateUtil.toLocalDateEuropeRome(request.getFromDate()),
                DateUtil.toLocalDateEuropeRome(request.getToDate()),
                request.getPagination()
        );

所以我不得不添加 .collect(Collectors.reducing((i1, i2) -> i1)) （我复制/粘贴了这个所以我只猜测它做了什么......它将 Flux 转换为 Mono），并转换我的 responseIds与（（ArrayList）responseIds.get（））。

repeat 不是最终的解决方案，因为它只重复 flatMap 中的内容（它不会重复连接到它的帖子）所以我不得不使用一个技巧......我删除了不必要的 for 循环，我在里面做了一个帖子我的 flatMap 与另一个 flatMap 重复......唯一缺少的是跟踪我的索引，我发现你可以使用 AtomicInteger 来做到这一点。这根本不是一件容易的事，但我测试了它并且它正在工作。回顾一下：

flatMapMany 里面有一个 repeat flatMap（repeat 只需要 long 作为一个参数，所以它会重复直到达到那个值并自动递增......但你不能使用这个索引，我理解）。
flatMap repeat 中的另一个 flatMap，这是因为如果没有此解决方法，您将无法进行另一个 post 调用，因为 repeat 只会重复 flatMap 内部的内容（不是之前的 post 调用，但它可以在其中进行 post 调用）。
一个 AtomicInteger 作为您的索引。
将返回类型更改为 Flux，收集并投射。
希望有人能从我的头痛中受益。

【讨论】：