如何通过Spring Webflux实现HTTP 2.0协议下的双向流式数据传输

结论 #

可以基于Spring Webflux + HTTP 2.0实现你要求的双向流式传输，全程无需收集全量数据即可边接收请求流边返回响应流。你之前的实现存在几处核心错误，修正后即可正常运行。

原有写法问题分析 #

• 写法1：@RequestBody声明为单个Integer类型，Spring会默认把整个请求体全量接收后解析为单个整数，完全不支持流式请求，逻辑也不符合需求。
• 写法2：一是缺少@PostMapping注解以及请求/响应媒体类型声明，二是代码存在笔误i.map(n -> String.valueOf(i))，把整个Flux对象转为字符串而非处理流中的元素。
• 写法3：@RequestBody声明为List<Integer>，Spring会全量接收所有请求数据转为列表后才开始处理，完全不符合流式处理要求。

正确实现代码 #

前提配置 #

首先确保服务端和客户端都开启HTTP2支持，Spring Boot项目在application.properties中添加配置即可：

server.http2.enabled=true
spring.webflux.codec.max-in-memory-size=0 # 关闭内存缓冲限制，避免数据量较大时被拦截攒量

服务端代码 #

import org.springframework.http.MediaType;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RestController;
import reactor.core.publisher.Flux;

@RestController
public class StreamController {

    @PostMapping(value = "/stream",
            // 指定流式请求的内容类型，使用换行分隔的ndjson格式，无需攒全量即可逐行解析
            consumes = MediaType.APPLICATION_NDJSON_VALUE,
            produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public Flux<String> stream(@RequestBody Flux<Integer> inputFlux) {
        // 直接对请求流中的元素做实时映射，无需等待全量数据
        return inputFlux.map(num -> "处理结果：" + num)
                .log();
    }
}

如果是Spring Boot 2.x版本，没有内置APPLICATION_NDJSON_VALUE常量，直接填字符串"application/x-ndjson"即可

客户端代码修正 #

核心是要指定请求的内容类型为ndjson，避免默认的JSON格式触发全量攒包：

import org.springframework.http.MediaType;
import org.springframework.web.reactive.function.BodyInserters;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import java.time.Duration;

public Mono<Void> stream() {
    var input = Flux.range(1, 10).delayElements(Duration.ofMillis(500));
    return stockWebClient.post()
            .uri("/stream")
            // 关键：指定流式请求的内容类型
            .contentType(MediaType.APPLICATION_NDJSON)
            .body(BodyInserters.fromPublisher(input, Integer.class))
            .accept(MediaType.TEXT_EVENT_STREAM)
            .retrieve()
            .bodyToFlux(String.class)
            .log()
            .then();
}

关键注意事项 #

• HTTP 2.0是实现真正全双工双向流的前提，HTTP 1.1仅支持半双工的流传输，无法做到请求和响应完全并行传输。
• 不要使用默认的application/json作为请求/响应的内容类型，JSON格式的解析依赖完整的结构，必然会触发全量数据攒包，要选择application/x-ndjson、text/event-stream这类流式友好的格式，每条数据可单独解析处理。

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