使用 RestTemplate 获取 InputStream

Question

我正在使用 URL 类从中读取 InputStream。有什么方法可以使用 RestTemplate 吗？

InputStream input = new URL(url).openStream();
JsonReader reader = new JsonReader(new InputStreamReader(input, StandardCharsets.UTF_8.displayName())); 

如何使用RestTemplate 而不是使用URL 获得InputStream？

Answer 1

我通过这样做来解决它。 希望对大家有帮助。

    @GetMapping("largeFile")
    public ResponseEntity<InputStreamResource> downloadLargeFile(
            @RequestParam("fileName") String fileName
    ) throws IOException {

        RestTemplate restTemplate = new RestTemplate();

        // Optional Accept header
        RequestCallback requestCallback = request -> request.getHeaders()
                .setAccept(Arrays.asList(MediaType.APPLICATION_OCTET_STREAM, MediaType.ALL));

        // Streams the response instead of loading it all in memory
        ResponseExtractor<InputStreamResource> responseExtractor = response -> {
            // Here I write the response to a file but do what you like
            Path path = Paths.get("tmp/" + fileName);
            Files.copy(response.getBody(), path, StandardCopyOption.REPLACE_EXISTING);
            return new InputStreamResource(new FileInputStream(String.format("tmp/%s", fileName)));
        };

        InputStreamResource response = restTemplate.execute(
            String.format("http://%s:%s/file/largeFileRestTemplate?fileName=%s", host, "9091", fileName),
            HttpMethod.GET,
            requestCallback,
            responseExtractor
        );

        return ResponseEntity
            .ok()
            .header(HttpHeaders.CONTENT_DISPOSITION, String.format("attachment; filename=%s", fileName))
            .body(response);
    }

Answer 2

您不应该直接获取InputStream。 RestTemplate 旨在封装处理响应（和请求）内容。它的优势在于处理所有 IO 并为您提供一个现成的 Java 对象。

RestTemplate 的原作者之一 Brian Clozel 拥有stated：

RestTemplate 并不意味着流式传输响应正文；它的合同 不允许，而且它已经存在了很长时间，以至于改变这种 它的行为的一个基本部分不能在不破坏许多的情况下完成 应用程序。

您需要注册适当的HttpMessageConverter 对象。这些将可以通过HttpInputMessage 对象访问响应的InputStream。

As Abdull suggests，Spring 确实为Resource 提供了一个HttpMessageConverter 实现，它本身包装了一个InputStream、ResourceHttpMessageConverter。它不支持所有的Resource 类型，但既然你应该对接口进行编程，你应该只使用超级接口Resource。

当前实现 (4.3.5) 将返回一个 ByteArrayResource，并将响应流的内容复制到您可以访问的新 ByteArrayInputStream。

您不必关闭流。 RestTemplate 会为您解决这个问题。 （如果您尝试使用InputStreamResource，这是很遗憾的，这是ResourceHttpMessageConverter 支持的另一种类型，因为它包装了底层响应的InputStream，但在它可以暴露给您的客户端代码之前被关闭。）

Answer 3

使用ResponseExtractor 是非常简单但有效的解决方案。当您想在非常大的 InputStream 上操作并且您的 RAM 有限时，它特别有用。

以下是你应该如何实现它：

public void consumerInputStreamWithoutBuffering(String url, Consumer<InputStream> streamConsumer) throws IOException {

    final ResponseExtractor responseExtractor =
            (ClientHttpResponse clientHttpResponse) -> {
                streamConsumer.accept(clientHttpResponse.getBody());
                return null;
            };

    restTemplate.execute(url, HttpMethod.GET, null, responseExtractor);
}

然后，在您需要的任何地方调用该方法：

Consumer<InputStream> doWhileDownloading = inputStream -> {
                //Use inputStream for your business logic...
};

consumerInputStreamWithoutBuffering("https://localhost.com/download", doWhileDownloading);

请注意以下常见陷阱：

public InputStream getInputStreamFromResponse(String url) throws IOException {

    final ResponseExtractor<InputStream> responseExtractor =
            clientHttpResponse -> clientHttpResponse.getBody();

    return restTemplate.execute(url, HttpMethod.GET, null, responseExtractor);
}

这里InputStream将在您访问之前关闭

Answer 4

我遇到了同样的问题，通过扩展 RestTemplate 并仅在读取流后关闭连接来解决。

您可以在此处查看代码：https://github.com/ItamarBenjamin/stream-rest-template

Answer 5

您可以传入自己的响应提取器。这是一个示例，我以流方式将 json 写入磁盘 -

        RestTemplate restTemplate = new RestTemplateBuilder().basicAuthentication("user", "their_password" ).build();

        int responseSize = restTemplate.execute(uri,
            HttpMethod.POST,
            (ClientHttpRequest requestCallback) -> {
                requestCallback.getHeaders().setContentType(MediaType.APPLICATION_JSON);
                requestCallback.getBody().write(body.getBytes());
            },
            responseExtractor -> {
                FileOutputStream fos  = new FileOutputStream(new File("out.json"));
                return StreamUtils.copy(responseExtractor.getBody(), fos);
            }
    )

Answer 6

前面的答案没有错，但没有深入到我喜欢看到的深度。在某些情况下，处理低级 InputStream 不仅是可取的，而且是必要的，最常见的例子是将大文件从源（某些 Web 服务器）流式传输到目标（数据库）。如果您尝试使用ByteArrayInputStream，您将毫不奇怪地收到OutOfMemoryError 的欢迎。是的，您可以滚动自己的 HTTP 客户端代码，但您必须处理错误的响应代码、响应转换器等。如果您已经在使用 Spring，那么寻找RestTemplate 是一个自然的选择。

在撰写本文时，spring-web:5.0.2.RELEASE 有一个ResourceHttpMessageConverter，它有一个boolean supportsReadStreaming，如果设置，并且响应类型为InputStreamResource，则返回InputStreamResource；否则返回ByteArrayResource。很明显，您不是唯一要求流媒体支持的人。

但是，有一个问题：RestTemplate 在 HttpMessageConverter 运行后不久关闭响应。因此，即使您要求InputStreamResource 并得到了它，这也不好，因为响应流已关闭。我认为这是他们忽略的设计缺陷；它应该取决于响应类型。所以不幸的是，为了阅读，你必须完全消耗响应；如果使用RestTemplate，则无法传递它。

不过写作没问题。如果您想流式传输InputStream，ResourceHttpMessageConverter 将为您完成。在底层，它使用org.springframework.util.StreamUtils 一次从InputStream 写入4096 个字节到OutputStream。

某些HttpMessageConverter 支持所有媒体类型，因此根据您的要求，您可能需要从RestTemplate 中删除默认的，并设置您需要的，注意它们的相对顺序。

最后但并非最不重要的一点是，ClientHttpRequestFactory 的实现有一个 boolean bufferRequestBody，如果您要上传大型流，您可以并且应该将其设置为 false。否则，你知道，OutOfMemoryError。在撰写本文时，SimpleClientHttpRequestFactory（JDK 客户端）和HttpComponentsClientHttpRequestFactory（Apache HTTP 客户端）支持此功能，但不支持OkHttp3ClientHttpRequestFactory。同样，设计监督。

编辑： 已提交工单SPR-16885。

Answer 7

感谢 Abhijit Sarkar 带路的回答。

我需要下载大量 JSON 流并将其分解为可流式管理的小型数据片段。 JSON 由具有大属性的对象组成：这些大属性可以序列化为文件，从而从未编组的 JSON 对象中删除。

另一个用例是逐个对象下载 JSON 流对象，像 map/reduce 算法一样处理它并生成单个输出，而无需将整个流加载到内存中。

另一个用例是读取一个大的 JSON 文件，然后根据条件只选择几个对象，同时解组为普通的旧 Java 对象。

这里有一个例子：我们想流式传输一个非常大的 JSON 文件，它是一个数组，并且我们想只检索数组中的第一个对象。

鉴于服务器上的这个大文件，可在http://example.org/testings.json 获得：

[
    { "property1": "value1", "property2": "value2", "property3": "value3" },
    { "property1": "value1", "property2": "value2", "property3": "value3" },
    ... 1446481 objects => a file of 104 MB => take quite long to download...
]

这个 JSON 数组的每一行都可以解析为这个对象：

@lombok.Data
public class Testing {
    String property1;
    String property2;
    String property3;
}

你需要这个类使解析代码可重用：

import com.fasterxml.jackson.core.JsonParser;
import java.io.IOException;
@FunctionalInterface
public interface JsonStreamer<R> {
    /**
     * Parse the given JSON stream, process it, and optionally return an object.<br>
     * The returned object can represent a downsized parsed version of the stream, or the result of a map/reduce processing, or null...
     *
     * @param jsonParser the parser to use while streaming JSON for processing
     * @return the optional result of the process (can be {@link Void} if processing returns nothing)
     * @throws IOException on streaming problem (you are also strongly encouraged to throw HttpMessageNotReadableException on parsing error)
     */
    R stream(JsonParser jsonParser) throws IOException;
}

还有这个类要解析：

import com.fasterxml.jackson.core.JsonFactory;
import com.fasterxml.jackson.core.JsonParser;
import lombok.AllArgsConstructor;
import org.springframework.http.HttpInputMessage;
import org.springframework.http.HttpOutputMessage;
import org.springframework.http.MediaType;
import org.springframework.http.converter.HttpMessageConverter;

import java.io.IOException;
import java.util.Collections;
import java.util.List;

@AllArgsConstructor
public class StreamingHttpMessageConverter<R> implements HttpMessageConverter<R> {

    private final JsonFactory factory;
    private final JsonStreamer<R> jsonStreamer;

    @Override
    public boolean canRead(Class<?> clazz, MediaType mediaType) {
        return MediaType.APPLICATION_JSON.isCompatibleWith(mediaType);
    }

    @Override
    public boolean canWrite(Class<?> clazz, MediaType mediaType) {
        return false; // We only support reading from an InputStream
    }

    @Override
    public List<MediaType> getSupportedMediaTypes() {
        return Collections.singletonList(MediaType.APPLICATION_JSON);
    }

    @Override
    public R read(Class<? extends R> clazz, HttpInputMessage inputMessage) throws IOException {
        try (InputStream inputStream = inputMessage.getBody();
             JsonParser parser = factory.createParser(inputStream)) {
            return jsonStreamer.stream(parser);
        }
    }

    @Override
    public void write(R result, MediaType contentType, HttpOutputMessage outputMessage) {
        throw new UnsupportedOperationException();
    }

}

然后，这里是用于流式传输 HTTP 响应、解析 JSON 数组并仅返回第一个未编组对象的代码：

// You should @Autowire these:
JsonFactory jsonFactory = new JsonFactory();
ObjectMapper objectMapper = new ObjectMapper();
RestTemplateBuilder restTemplateBuilder = new RestTemplateBuilder();

// If detectRequestFactory true (default): HttpComponentsClientHttpRequestFactory will be used and it will consume the entire HTTP response, even if we close the stream early
// If detectRequestFactory false: SimpleClientHttpRequestFactory will be used and it will close the connection as soon as we ask it to

RestTemplate restTemplate = restTemplateBuilder.detectRequestFactory(false).messageConverters(
    new StreamingHttpMessageConverter<>(jsonFactory, jsonParser -> {

        // While you use a low-level JsonParser to not load everything in memory at once,
        // you can still profit from smaller object mapping with the ObjectMapper
        if (!jsonParser.isClosed() && jsonParser.nextToken() == JsonToken.START_ARRAY) {
            if (!jsonParser.isClosed() && jsonParser.nextToken() == JsonToken.START_OBJECT) {
                return objectMapper.readValue(jsonParser, Testing.class);
            }
        }
        return null;

    })
).build();

final Testing firstTesting = restTemplate.getForObject("http://example.org/testings.json", Testing.class);
log.debug("First testing object: {}", firstTesting);

Answer 8

Spring 有一个org.springframework.http.converter.ResourceHttpMessageConverter。它转换 Spring 的 org.springframework.core.io.Resource 类。 那个Resource类封装了一个InputStream，可以通过someResource.getInputStream()获取。

综上所述，您实际上可以通过将Resource.class 指定为RestTemplate 调用的响应类型，通过RestTemplate 开箱即用地获得InputStream。

这是使用RestTemplate 的exchange(..) 方法之一的示例：

import org.springframework.web.client.RestTemplate;
import org.springframework.http.HttpMethod;
import org.springframework.core.io.Resource;

ResponseEntity<Resource> responseEntity = restTemplate.exchange( someUrlString, HttpMethod.GET, someHttpEntity, Resource.class );

InputStream responseInputStream;
try {
    responseInputStream = responseEntity.getBody().getInputStream();
}
catch (IOException e) {
    throw new RuntimeException(e);
}

// use responseInputStream

Answer 9

作为一种变体，您可以将响应作为字节使用，然后转换为流

byte data[] = restTemplate.execute(link, HttpMethod.GET, null, new BinaryFileExtractor());
return new ByteArrayInputStream(data);

提取器是

public class BinaryFileExtractor implements ResponseExtractor<byte[]> {

  @Override
  public byte[] extractData(ClientHttpResponse response) throws IOException {
    return ByteStreams.toByteArray(response.getBody());
  }
}