java - 如何解压缩 Flux (以及如何编写)？

Question

我需要在没有中间存储的情况下读取和写入压缩 (GZIP) 流。目前，我正在使用 Spring RestTemplate进行写入，Apache HTTP 客户端进行读取(请参阅我的回答 here 以了解为什么 RestTemplate 不能用于读取大型流的解释)。实现相当简单，我在其中打了一个GZIPInputStream。关于回应 InputStream继续前进。

现在，我想切换到使用 Spring 5 WebClient (只是因为我不喜欢现状)。然而，WebClient本质上是 react 性的，处理 Flux<Stuff> ;我相信有可能得到 Flux<DataBuffer> , 其中DataBuffer是对 ByteBuffer 的抽象.问题是，我如何即时解压缩它而不必将完整流存储在内存中( OutOfMemoryError ，我正在看着你)或写入本地磁盘？值得一提的是 WebClient在引擎盖下使用 Netty。

• 另见 Reactor Netty issue-251 .
• 也与 Spring 集成有关 issue-2300 .

我承认我对(解)压缩知之甚少，但是，我做了我的研究，但网上提供的资料似乎都没有特别有用。

compression on java nio direct buffers

Writing GZIP file with nio

Reading a GZIP file from a FileChannel (Java NIO)

(de)compressing files using NIO

Iterable gzip deflate/inflate in Java

Answer 1

public class HttpResponseHeadersHandler extends ChannelInboundHandlerAdapter {
    private final HttpHeaders httpHeaders;

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        if (msg instanceof HttpResponse &&
                !HttpStatus.resolve(((HttpResponse) msg).status().code()).is1xxInformational()) {
            HttpHeaders headers = ((HttpResponse) msg).headers();

            httpHeaders.forEach(e -> {
                log.warn("Modifying {} from: {} to: {}.", e.getKey(), headers.get(e.getKey()), e.getValue());
                headers.set(e.getKey(), e.getValue());
            });
        }
        ctx.fireChannelRead(msg);
    }
}

然后我创建一个 ClientHttpConnector 与 WebClient 一起使用，并在 afterNettyContextInit 中添加处理程序:

ctx.addHandlerLast(new ReadTimeoutHandler(readTimeoutMillis, TimeUnit.MILLISECONDS));
ctx.addHandlerLast(new Slf4JLoggingHandler());
if (forceDecompression) {
    io.netty.handler.codec.http.HttpHeaders httpHeaders = new ReadOnlyHttpHeaders(
            true,
            CONTENT_ENCODING, GZIP,
            CONTENT_TYPE, APPLICATION_JSON
    );
    HttpResponseHeadersHandler headersModifier = new HttpResponseHeadersHandler(httpHeaders);
    ctx.addHandlerFirst(headersModifier);
}
ctx.addHandlerLast(new HttpContentDecompressor());

当然，对于未进行 GZIP 压缩的响应，这将失败，因此我仅将此 WebClient 实例用于特定用例，我确定响应已压缩。

编写很简单:Spring 有一个ResourceEncoder，所以InputStream 可以简单地转换为InputStreamResource，瞧!