使用 Spring Reactive WebClient 同时流式传输请求/响应主体？答案

【问题标题】：Simultaneously streaming request / response bodies with the Spring Reactive WebClient?使用 Spring Reactive WebClient 同时流式传输请求/响应主体？
【发布时间】：2020-03-29 09:33:16
【问题描述】：

我在以下客户端-服务器交互中看到一些奇怪的行为，我想知道我是否遇到了 HTTP/1.1 语义或我的反应式编程技能需要工作（或两者兼而有之）。

我正在尝试创建客户端-服务器交互，其中请求和响应主体都是长时间运行的数据流。

客户端是一个 Spring Reactive WebClient，它在请求正文中发送无限流。它期望接收（并记录）无限的结果流。

Flux<Long> requests = Flux.interval(Duration.ofSeconds(2));

return WebClient.create()
        .post()
            .uri("/instructions")
            .contentType(MediaType.APPLICATION_STREAM_JSON)
            .body(requests, Long.class)
        .retrieve()
            .bodyToFlux(Object.class)
            .map(response -> {
                log.info("Received Response Object {}", response);
                return response;
            });

服务器是一个spring-boot-starter-webflux 应用程序，带有一个路由处理程序，用于在收到请求对象时记录它们并提供无限的结果流：

public Mono<ServerResponse> instructions(ServerRequest request) {
    // Log the request objects as they are received
    Flux<Object> requestStream = request.bodyToFlux(Object.class)
            .map(r -> {
                log.info("Received Request Object: {}", r);
                return r;
            });

    requestStream.subscribe();

    // Infinite stream of responses
    Flux<Long> responses = Flux.interval(Duration.ofSeconds(5));

    return ServerResponse.ok()
            .contentType(MediaType.APPLICATION_STREAM_JSON)
            .body(responses, Long.class);
}

当上面的代码运行时，服务器会记录无限的请求对象流，但客户端从不记录任何响应对象。

如果我通过以下方式绑定请求流：Flux<Long> requests = Flux.interval(Duration.ofSeconds(2)).take(20);，那么客户端会在收到所有请求后开始记录响应。

这里有什么问题？ * 响应式代码有问题吗？ * 这部分 HTTP/1.1 规范是否在完全接收到请求正文之前不应发送响应标头？

【问题讨论】：

我不认为规范有什么问题，但我尝试过类似的事情，有些（很多？）HTTP服务器和客户端不会同时处理流请求和响应对象很好。

标签： java spring reactive-programming spring-webflux

【解决方案1】：

看到这个问题：WebClient doesn't read response until request write is completed.

显然，基于 Netty 的 WebClient 只有在发送完请求后才开始处理响应

Netty 服务器的行为也可能类似（仅在读取完请求正文后才开始发送响应），我不确定

有趣的是，基于 Jetty 的 WebClient 能够同时处理请求和响应

【讨论】：