java - FlatMap a Flux 未执行

Question

我有一个包含 240 个项目的列表，使用 for 完整发送该列表需要 1 个多小时。

List<Map<String, Object>> conventions = mapConventions(objects, referentialService);

for(Map<String, Object> item : conventions) {
        webClient.post()
        .uri(configProperties.getUrl().getConvention() + CONVENTION)
        .bodyValue(objectMapper.convertValue(item, JsonNode.class))
        .retrieve()
        .bodyToMono(String.class);
}

所以我遵循了这个article同时发送以最大限度地缩短响应时间，但 flatmap 内的代码永远不会执行:

Flux.fromIterable(conventions).flatMap(item ->  {
    System.out.print(item);
    return webClient.post()
            .uri(configProperties.getUrl().getConvention() + CONVENTION)
            .bodyValue(objectMapper.convertValue(item, JsonNode.class))
            .retrieve()
            .bodyToMono(String.class);
});

Answer 1

在响应式编程中，有生产者和订阅者。虽然生产者可以输出结果，但如果没有任何东西在监听这些结果，那么它就没有任何好处——这就是订阅者发挥作用的地方。订阅者处理生产者的输出，并对结果执行一些有意义的操作。这是响应式(Reactive)编程的基础，如果订阅者没有“监听”结果，生产者将不会执行任何代码。

因此，在本例中，flatmap() 是生产者。除非有订阅者来处理输出，否则它不会执行任何代码。

简短的答案是在平面图末尾添加 subscribe() 调用。看起来像这样。

Flux.fromIterable(conventions).flatMap(item ->  {
    System.out.print(item);
    return webClient.post()
            .uri(configProperties.getUrl().getConvention() + CONVENTION)
            .bodyValue(objectMapper.convertValue(item, JsonNode.class))
            .retrieve()
            .bodyToMono(String.class);
}).subscribe();

许多教程都是关于此的。 例如: https://spring.io/blog/2016/06/13/notes-on-reactive-programming-part-ii-writing-some-code https://medium.com/@olehdokuka/mastering-own-reactive-streams-implementation-part-1-publisher-e8eaf928a78c https://projectreactor.io/docs/core/release/reference/