实际上这是个好问题,因为要理解WebFlux或者Project Reactor框架,当需要链接微服务时需要几个步骤。
第一步是意识到WebClient应该接受一个发布者并返回一个发布者。将其推广到4个不同的方法签名来帮助思考。
- Mono -> Mono
- Flux -> Flux
- Mono -> Flux
- Flux -> Mono
在所有情况下,无疑都是Publisher->Publisher, 但在你更好地理解之前,暂且只考虑前两种。你只需使用".map(...)"来处理流中的对象即可,但你需要学习如何处理后两个。正如上面所评论的,从Flux -> Mono可以使用".collectList()"或".reduce(...)"完成。而从Mono -> Flux似乎通常使用".flatMapMany"或".flatMapIterable"或其某种变体。可能还有其他技巧。在任何WebFlux代码中都不应使用".block()",如果尝试这样做,通常会收到运行时错误。
在你的示例中,你想要转换成
(Mono -> Flux) -> (Flux -> Flux) -> (Flux -> Flux)
就像你说的那样,你想要
Mono -> Flux -> Flux
第二部分是了解如何链接流(chaining Flows)。你可以这样做
p3(p2(p1(object)));
这将链式调用p1 -> p2 -> p3,但我总觉得将其转化为"Service Layer"更容易理解。
o2 = p1(object);
o3 = p2(o2);
result = p3(o3);
这段代码更容易阅读和维护,随着一些成熟的经验,你会明白这种说法的价值。
我在您的示例中遇到的唯一问题是使用
WebClient
作为
@RequestBody
来执行
Flux<String>
。这不起作用。请参见
WebClient bodyToFlux(String.class) for string list doesn't separate individual values。除此之外,这是一个非常直接的应用程序。当您调试它时,您会发现它在执行
Flux<Integer> ids = mapNamesToIds(fn)
行之前就已经到达了
.subscribe(System.out::println)
行。这是因为在执行之前设置了流量。需要花些时间理解,但这是项目反应器框架的重点。
@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()));
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;
}
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);
}
@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;
}