平面映射未执行的通量

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);
}

所以我跟着这篇文章同时发送，以尽量减少响应时间，但是平板地图里面的代码永远不会执行：

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

在反应式编程中，有生产者和订阅者。虽然制作人可以输出结果，但如果没有人在听这些结果，那就没有任何好处——订阅者就是在听这些结果。订阅者处理生产者的输出，并对结果进行有意义的处理。对于反应式编程来说，如果订阅者没有“监听”结果，那么生产者就不会执行任何代码。

所以在这种情况下，flatmap（）是生产者。并且它不会执行任何代码，除非有订阅服务器来处理输出。

简单的回答是在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-e8eaf928a78chttps://projectreactor.io/docs/core/release/reference/