Spring WebFlux微服务间发送Integer类型Flux报错求助

我是Spring WebFlux和WebClient的新手，正在做POC，设计了两个微服务。尝试从服务A通过POST请求发送Integer类型的Flux，在服务B中接收，但遇到了错误。

服务A代码 #

Flux<Integer> integerFlux = Flux.just(1,2,3,4,5);
WebClient.create("http://localhost:9002/registry")
         .post()
         .uri("/packages")
         .contentType(MediaType.APPLICATION_JSON)
         .body(integerFlux, Integer.class)
         .retrieve()
         .bodyToMono(Void.class)
         .block();

服务B代码 #

@PostMapping(value = "/packages", consumes = MediaType.APPLICATION_JSON_VALUE)
public ResponseEntity<Void> createPackage(@RequestBody Flux<Integer> integerFlux) {
    this.registryService.createPackage(integerFlux);
    return new ResponseEntity<>(HttpStatus.CREATED);
}

错误日志 #

服务A错误日志 #

org.springframework.web.reactive.function.client.WebClientResponseException$InternalServerError: 500 Internal Server Error from POST http://localhost:9002/registry/packages
    at org.springframework.web.reactive.function.client.WebClientResponseException.create(WebClientResponseException.java:239) ~[spring-webflux-5.3.24.jar:5.3.24]
    Suppressed: reactor.core.publisher.FluxOnAssembly$OnAssemblyException: 
Error has been observed at the following site(s):
    *__checkpoint ⇢ 500 from POST http://localhost:9002/registry/packages [DefaultWebClient]

服务B错误日志 #

Servlet.service() for servlet [dispatcherServlet] in context with path [] threw exception [Request processing failed; nested exception is org.springframework.http.converter.HttpMessageConversionException: Type definition error: [simple type, class reactor.core.publisher.Flux]; nested exception is com.fasterxml.jackson.databind.exc.InvalidDefinitionException: Cannot construct instance of `reactor.core.publisher.Flux` (no Creators, like default constructor, exist): abstract types either need to be mapped to concrete types, have custom deserializer, or contain additional type information
 at [Source: (org.springframework.util.StreamUtils$NonClosingInputStream); line: 1, column: 1]] with root cause

com.fasterxml.jackson.databind.exc.InvalidDefinitionException: Cannot construct instance of `reactor.core.publisher.Flux` (no Creators, like default constructor, exist): abstract types either need to be mapped to concrete types, have custom deserializer, or contain additional type information

尝试过的操作 #

我曾将Content-Type改为application/nd-json，但出现415 Unsupported Media Type错误，尝试相关解决方案也无效。

核心问题分析 #

服务B报错的本质原因：

1. 如果服务B是Spring MVC项目而非WebFlux，Jackson会尝试直接反序列化Flux抽象类，而Flux没有可用的构造方法，导致序列化失败。
2. 即使服务B是WebFlux，用application/json发送Flux时，WebClient会把整个Flux序列化为单个JSON数组，但@RequestBody Flux<Integer>期望的是流式的JSON元素（如ndjson格式），而非单个数组，因此无法匹配。

正确实现方式 #

方式一：使用NDJSON流式传输（推荐真正的流式场景）

1. 确保服务B是WebFlux项目：依赖spring-boot-starter-webflux，移除spring-boot-starter-web依赖，避免使用Spring MVC的DispatcherServlet。
2. 调整服务A代码：指定Content-Type为application/ndjson：

Flux<Integer> integerFlux = Flux.just(1,2,3,4,5);
WebClient.create("http://localhost:9002/registry")
         .post()
         .uri("/packages")
         .contentType(MediaType.APPLICATION_NDJSON)
         .body(integerFlux, Integer.class)
         .retrieve()
         .bodyToMono(Void.class)
         .block();

3. 调整服务B接口：修改consumes为对应格式：

@PostMapping(value = "/packages", consumes = MediaType.APPLICATION_NDJSON_VALUE)
public ResponseEntity<Void> createPackage(@RequestBody Flux<Integer> integerFlux) {
    this.registryService.createPackage(integerFlux);
    return new ResponseEntity<>(HttpStatus.CREATED);
}

方式二：发送JSON数组并转为Flux（兼容性好，非纯流式）

如果不需要纯流式传输，可以将Flux转为List后发送，WebFlux会自动将接收的JSON数组拆分为Flux：

1. 服务A代码：

Flux<Integer> integerFlux = Flux.just(1,2,3,4,5);
WebClient.create("http://localhost:9002/registry")
         .post()
         .uri("/packages")
         .contentType(MediaType.APPLICATION_JSON)
         .body(integerFlux.collectList(), List.class)
         .retrieve()
         .bodyToMono(Void.class)
         .block();

2. 服务B代码：保持原application/json配置即可：

@PostMapping(value = "/packages", consumes = MediaType.APPLICATION_JSON_VALUE)
public ResponseEntity<Void> createPackage(@RequestBody Flux<Integer> integerFlux) {
    this.registryService.createPackage(integerFlux);
    return new ResponseEntity<>(HttpStatus.CREATED);
}

内容的提问来源于stack exchange，提问作者Ronit Nath