如何在订阅中访问 Flux 发出的值

Question

Flux 发出的项目（在本例中为“Red”、“White”、“Blue”）被传递给外部服务调用。我从returnValue 中的外部服务获取响应值。如何将发送到外部服务的元素与收到的响应进行映射？

@Log4j2
@SpringBootApplication
class FluxFromIterableAccessFlatMapValue implements CommandLineRunner {

    @Override
    public void run(String... args) throws Exception {

        Flux.just("Red", "White", "Blue").
                flatMap(colors -> Mono.fromCallable(() -> {
                    // > Call to external service made here.
                    return "Return value from external Service call.";
                })).subscribeOn(Schedulers.single())
                .subscribe(returnValue ->
                        log.info("Need to access which element produced this response?" +
                                "Is it response for Red, White or Blue? " + returnValue));

    }
}

Answer 1

我会简单地使用元组（或任何其他包装器）将每个响应与相应的颜色配对，如下所示：

Mono<Tuple2<String, String>> makeExternalCall(String color) {
     return Mono.fromCallable(() -> {
            // > Call to external service made here.
            return "Return value from external Service call for color: " + color;
        })
        .map(response -> Tuples.of(color, response));
}

Flux.just("Red", "White", "Blue")
    .flatMap(this::makeExternalCall)//Flux<Tuple2<String, String>>
    .subscribeOn(Schedulers.single())
    .subscribe(returnValue -> log.info("Is it response for Red, White or Blue? " + returnValue));

示例响应：

Is it response for Red, White or Blue? [Red,Return value from external Service call for color:  Red]
Is it response for Red, White or Blue? [White,Return value from external Service call for color:  White]
Is it response for Red, White or Blue? [Blue,Return value from external Service call for color:  Blue]