反应式编程：SpringWebFlux：如何构建微服务调用链？

Question

Spring Boot应用程序：

@RestController接收以下有效负载：

{
  "cartoon": "The Little Mermaid",
  "characterNames": ["Ariel", "Prince Eric", "Sebastian", "Flounder"]
}

我需要按以下方式进行处理：

>

使用转换后的数据向“卡通数据库”微服务发送HTTP POST请求。

我遇到的问题是：

我需要使用Spring-WebFlux（Mono|Flux）和Spring-Reactive-WebClient）的反应式编程（非阻塞\异步处理）范例来实现所有这些步骤，但我对该堆栈没有任何经验，我尽可能多地阅读它，加上谷歌搜索的次数很多，但仍然有一大堆未回答的问题，例如：

Q1.我已经配置了反应性webClient，它向“卡通人物”微服务发送请求：

      public Mono<Integer> getCartoonCharacterIdbyName(String characterName) {
    return WebClient.builder().baseUrl("http://cartoon-characters").build()
        .get()
        .uri("/character/{characterName}", characterName)
        .retrieve()
        .bodyToMono(Integer.class);
  }

正如你可能看到的，我有一个列表的卡通人物的名字和他们中的每一个我需要调用getCartoon特性IdbyName（字符串名称）方法，我不确定正确的选项来调用它串联，相信正确的选项：并行执行。

写了以下方法：

  public List<Integer> getCartoonCharacterIds(List<String> names) {
Flux<Integer> flux = Flux.fromStream(names.stream())
    .flatMap(this::getCartoonCharacterIdbyName);

return StreamSupport.stream(flux.toIterable().spliterator(), false)
    .collect(Collectors.toList());

}

但我怀疑，这段代码是否并行WebClient执行，以及代码调用flux。toIterable（）阻塞线程，因此在这个实现中，我丢失了非阻塞机制。

我的假设正确吗？

我需要如何将其重写为具有并行性和非阻塞性？

Q2.在技术上是否可以将控制器接收的输入数据（我的意思是用id替换名称）以反应性方式转换：当我们使用Flux操作时

第三季度。是否有可能不仅获取转换后的数据对象，而且获取Mono




Answer 1

实际上，这是一个很好的问题，因为当涉及到链接微服务时，理解WebFlux或项目反应器框架需要几个步骤。

第一个是认识到WebClient应该接收发布者并返回发布者。将其外推到4种不同的方法特征，以帮助思考。

• 单声道-

当然，在所有情况下，它只是Publisher-

在您的示例中，您希望转到

• （单声道）-

正如你所说，你想要

• 单声道-

第二部分是了解链接流。你可以

• p3（p2（p1（对象））

这将链p1-

• o2=p1（对象）
• o3=p2（o2）
• 结果=p3（o3）

这段代码更易于阅读和维护，并且随着一些成熟度的提高，您逐渐理解了这段语句的价值。

我对您的示例的唯一问题是执行通量

@SpringBootApplication
@RestController
@RequestMapping("/demo")
public class DemoApplication implements ApplicationRunner {

    public static void main(String[] args) {
        SpringApplication.run(DemoApplication.class, args);
    }

    Map<Integer, CartoonCharacter> characters;

    @Override
    public void run(ApplicationArguments args) throws Exception {
        String[] names = new String[] {"Ariel", "Prince Eric", "Sebastian", "Flounder"};
        characters = Arrays.asList( new CartoonCharacter[] {
                new CartoonCharacter(names[0].hashCode(), names[0], "Mermaid"), 
                new CartoonCharacter(names[1].hashCode(), names[1], "Human"), 
                new CartoonCharacter(names[2].hashCode(), names[2], "Crustacean"), 
                new CartoonCharacter(names[3].hashCode(), names[3], "Fish")} 
        )
        .stream().collect(Collectors.toMap(CartoonCharacter::getId, Function.identity()));
        // TODO Auto-generated method stub
        CartoonRequest cr = CartoonRequest.builder()
        .cartoon("The Little Mermaid")
        .characterNames(Arrays.asList(names))
        .build();
        thisLocalClient
            .post()
            .uri("cartoonDetails")
            .body(Mono.just(cr), CartoonRequest.class)
            .retrieve()
            .bodyToFlux(CartoonCharacter.class)
            .subscribe(System.out::println);
    }

    @Bean
    WebClient localClient() {
        return WebClient.create("http://localhost:8080/demo/");
    }

    @Autowired
    WebClient thisLocalClient;

    @PostMapping("cartoonDetails")
    Flux<CartoonCharacter> getDetails(@RequestBody Mono<CartoonRequest> cartoonRequest) {
        Flux<StringWrapper> fn = cartoonRequest.flatMapIterable(cr->cr.getCharacterNames().stream().map(StringWrapper::new).collect(Collectors.toList()));
        Flux<Integer> ids = mapNamesToIds(fn);
        Flux<CartoonCharacter> details = mapIdsToDetails(ids);
        return details;
    }
    //  Service Layer Methods
    private Flux<Integer> mapNamesToIds(Flux<StringWrapper> names) {
        return thisLocalClient
            .post()
            .uri("findIds")
            .body(names, StringWrapper.class)
            .retrieve()
            .bodyToFlux(Integer.class);
    }
    private Flux<CartoonCharacter> mapIdsToDetails(Flux<Integer> ids) {
        return thisLocalClient
            .post()
            .uri("findDetails")
            .body(ids, Integer.class)
            .retrieve()
            .bodyToFlux(CartoonCharacter.class);
    }
    // Services
    @PostMapping("findIds")
    Flux<Integer> getIds(@RequestBody Flux<StringWrapper> names) {
        return names.map(name->name.getString().hashCode());
    }
    @PostMapping("findDetails")
    Flux<CartoonCharacter> getDetails(@RequestBody Flux<Integer> ids) {
        return ids.map(characters::get);
    }
}

也：

@Data
@NoArgsConstructor
@AllArgsConstructor
@Builder
public class StringWrapper {
    private String string;
}
@Data
@Builder
public class CartoonRequest {
    private String cartoon;
    private List<String> characterNames;
}
@Data
@Builder
@NoArgsConstructor
@AllArgsConstructor
public class CartoonCharacter {
    Integer id;
    String name;
    String species;
}