以 REST 下载方式流式传输资源

Question

因此，我有一个 REST 端点，用于从前端后端 (BFF) 下载文件：

@Path("/attachments")
public interface AttachmentApi {
    @GET
    @Path("/{id}")
    @Produces({"*/*", "application/problem+json"})
    Response getAttachment(@PathParam("id") String id);
}

然后，BFF 只是将请求转发到另一个后端，下载文件并将其转发到客户端。 这对于小文件来说效果很好，但对于大文件来说就不行了，因为它遇到了

DataBufferLimitException

s:

nested exception is org.springframework.core.io.buffer.DataBufferLimitException: Exceeded limit on max bytes to buffer : 262144.

是的，我可以简单地增加 Spring Boot 配置中的最大内存。但我发现它更适合流而不是下载文件，特别是对于较大的文件。因此，我调整了我的实施并提出了以下建议：

@Component
@Api
@PreAuthorize("hasAuthority('USER')")
@RequiredArgsConstructor
public class AttachmentApiImpl implements AttachmentApi {
    private static final Logger LOGGER = LoggerFactory.getLogger(AttachmentApiImpl.class);

    @Override
    public Response getAttachment(String id) {
        ResponseEntity<InputStreamResource> attachmentServiceResponse = findById(id);

        StreamingOutput streamingOutput = output -> {
            try (InputStream inputStream = Objects.requireNonNull(attachmentServiceResponse.getBody()).getInputStream()) {
                byte[] buffer = new byte[4096];
                int bytesRead;
                while ((bytesRead = inputStream.read(buffer)) != -1) {
                    output.write(buffer, 0, bytesRead);
                }
            } catch (IOException e) {
                throw new WebApplicationException("Error streaming file", e);
            }
        };

        return Response.ok(streamingOutput)
                .header(HttpHeaders.ACCESS_CONTROL_EXPOSE_HEADERS, HttpHeaders.ETAG + ", " + HttpHeaders.CONTENT_DISPOSITION)
                .header(HttpHeaders.ETAG, attachmentServiceResponse.getHeaders().getETag())
                .header(HttpHeaders.CONTENT_DISPOSITION, attachmentServiceResponse.getHeaders().getContentDisposition())
                .header(HttpHeaders.CONTENT_TYPE, attachmentServiceResponse.getHeaders().getContentType())
                .build();
    }

    private ResponseEntity<InputStreamResource> findById(String id) {
        return webClient.get()
                .uri(uriBuilder -> uriBuilder
                        .path("/attachments/{id}")
                        .build(id))
                .retrieve()
                .toEntity(InputStreamResource.class)
                .block();
    }
}

但这并没有真正改变任何事情。我仍然得到

DataBufferLimitException

，而且似乎它并没有真正按预期进行流式传输。我还阅读了一些有关使用

DataBuffer

和

DataBufferUtils

的内容，但我不太确定如何使用它。

有人知道我的代码有什么问题吗？

Answer 1

主要问题是你试图用

InputStream

和

OutputStream

来架起反应式和阻塞式世界的桥梁。这是行不通的，您当前的操作方式会将所有内容检索到

byte[]

中，并用

InputStream

包裹它。

如果您想要流式传输，则需要使用

DataBuffer

并将读取的数据复制到

OutputStream

。你不应该阻塞任何地方，而只能

subscribe

到最后的单声道。

private Mono<ResponseEntity<Flux<DataBuffer>> findById(String id) {
  return webClient.get()
      .uri(uriBuilder -> uriBuilder
          .path("/attachments/{id}")
          .build(id))
       .retrieve()
       .toEntityFlux(DataBuffer.class);
}

现在您将获得 1 个或多个

DatBuffer

，您可以使用

OutputStream

直接将其写入

DataBufferUtils

。

    @Override
    public Response getAttachment(String id) {
        Mono<ResponseEntity<Flux<DataBuffer>> attachmentServiceResponse = findById(id);

        StreamingOutput streamingOutput = output -> {
           attachmentServiceResponse.map( (res) -> DataBufferUtils.write(res.getBody(), output)).subscribe();
        };

        return Response.ok(streamingOutput);
}

缺点没有简单的方法来复制标题。如果您使用 Spring Webflux（或 Spring MVC）作为您的 Web 技术，您只需从方法中返回

Mono<ResponseEntity<Flux<DataBuffer>>

，Spring 就会为您处理一切。