POST InputStream 与 RestTemplate 和处理背压答案

【问题标题】：POST InputStream with RestTemplate and handling backpresurePOST InputStream 与 RestTemplate 和处理背压
【发布时间】：2021-08-15 13:41:18
【问题描述】：

我们有 3 层系统（UI、Server1、Server2）

我们必须将文件从我们的客户端（Angular）传输到某个服务器（java server1，Spring），这个服务器必须将文件数据传递给第二个服务器

UI => 服务器 1 ==> 服务器 2

我们希望将字节从 UI 流式传输到“服务器 2”，而不将文件存储在“服务器 1”的磁盘上或不将文件数据存储在“服务器 1”的内存中

链接“UI”==>“Server 1”快的问题

“服务器 1”==>“服务器 2”的链接很慢

我们面临的问题是文件数据在输出流中被缓冲到“服务器 2”

这是我们目前使用的函数：

private ResponseEntity<String> upload(String url, HttpHeaders headers, HttpServletRequest request) {
        // copy bytes from caller input stream to FE request stream
        RequestCallback requestCallback = (ClientHttpRequest req) -> {
            Utils.propagateHeaders(headers,req);
            request.getInputStream().transferTo(req.getBody());
        };

        // execute the call to server 2
        ResponseEntity executeResult = template.execute(url, HttpMethod.PUT, requestCallback, (clientHttpResponse) -> {
            HttpStatus statusCode = clientHttpResponse.getStatusCode();               
            return ResponseEntity.status(statusCode);
        });
        return executeResult;
    }

所以我们知道这是一个背压问题，request.getInputStream() 必须以输出流可以通过线路发送数据的速度将数据发送到输出流

最好的方法是什么？

谢谢

【问题讨论】：

标签： java stream backpressure

【解决方案1】：

最终替换了与接收的 WebClient 一起使用的 RestTemplate Flux 作为 API 调用主体的参数

// create a Flux of DataBuffer from the InputStream that we get from UI
   Flux<DataBuffer> dataBufferFlux = DataBufferUtils.readByteChannel(() -> Channels.newChannel(request.getInputStream()), new DefaultDataBufferFactory(), 8192);

   // call PUT API providing the dataBufferFlux as parameter for body
   Mono<ClientResponse> responseMono = webClient.put()
                .uri(url)
                .body(BodyInserters.fromDataBuffers(dataBufferFlux))
                .headers((HttpHeaders head) ->head.addAll(headers))
                .exchange();

此更改似乎减少了内存占用，并且并非所有内存都在调用 API 后立即分配

这是我必须添加的依赖项

<dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-webflux</artifactId>
</dependency>

【讨论】：