使用 spring webflux 使用第一次调用结果的多个异步调用

Question

我需要进行异步调用并使用其中存在的一些值对同一服务进行多次调用。将这些调用的响应与第一个调用和 return.

结合起来

例如，当我进行第一次调用时，我得到下面的 JSON，其中有一个 ID 列表。现在我必须使用这些 ID 多次调用服务并列出它们的响应并通过将其附加到相同的 JSON.

中将其发送到下游

 {“id”: 145,
    “object”:[{“id”:111}]
    }

我试过使用 zipWhen 和

Flux.fromIterable(userIds).parallel().runOn(Schedulers.elastic()).flatMap()

但是结果列表总是空的或空的。我们怎样才能做到这一点？我在这里遗漏了什么吗？

编辑 1: 使用 Flux.fromIterable 解决了它。阅读更多有关它的信息，终于了解了使用并解决了它。下面的方法从列表中获取一项并将调用内部方法，该方法将调用多个 APIs:

return Flux.fromIterable(somelist).flatMap(listItem -> {
    return someMethodToCallAnotherAPIUsingZipWith(listItem);
    }).collectList();

内在方法: 它调用第一个 API，将其结果传递给 zipWith 并使用此结果我们可以调用另一个 API 或者我们可以简单地将它与它的响应一起使用。

private Mono<Object> someMethodToCallAnotherAPIUsingZipWith(String listItem) {
        return authService.getAccessToken().flatMap(accessToken ->
                webClient.get().uri(builder -> builder.path("/path").build(listItem))
                        .header(HttpHeaders.AUTHORIZATION, accessToken)
                        .retrieve()
                        .toEntity(Entity.class).log()
                        .flatMap(entity -> {
                            //manipulate your response or create new object using it
                            return Mono.just(entity);
                        }).zipWhen(consent -> webClient.get().uri(builder -> builder.path("/otherpath").build(listItem))
                        .header(HttpHeaders.AUTHORIZATION, accessToken)
                        .retrieve().bodyToMono(Entity.class).log()
                        .flatMap(entity -> {
                            //example
                            listItem = entity.getString();
                            return Mono.just(listItem);
                        }), (string1, string2) -> string1 + string2));
    }

Answer 1

private Mono<Object> someMethodToCallAnotherAPIUsingZipWith(String listItem) {
    return authService.getAccessToken().flatMap(accessToken ->
            webClient.get().uri(builder -> builder.path("/path").build(listItem))
                    .header(HttpHeaders.AUTHORIZATION, accessToken)
                    .retrieve()
                    .toEntity(Entity.class).log()
                    .flatMap(entity -> {
                        //manipulate your response or create new object using it
                        return Mono.just(entity);
                    }).zipWhen(consent -> webClient.get().uri(builder -> builder.path("/otherpath").build(listItem))
                    .header(HttpHeaders.AUTHORIZATION, accessToken)
                    .retrieve().bodyToMono(Entity.class).log()
                    .flatMap(entity -> {
                        //example
                        listItem = entity.getString();
                        return Mono.just(listItem);
                    }), (string1, string2) -> string1 + string2));
}

使用 spring webflux 使用第一次调用结果的多个异步调用

Multiple asynchronous calls using result of first call using spring webflux

java

reactive-programming

spring-boot

spring-webflux

webflux