Java顺序解压GZIP流

Question

我的 Java 程序实现了一个服务器，它应该从客户端通过 websockets 获取一个非常大的文件，使用 gzip 压缩，并且应该检查文件内容中的某些字节模式。

客户端发送嵌入在专有协议中的文件块，因此我从客户端收到一条又一条消息，解析消息并提取压缩后的文件内容。

我无法将整个文件保存在程序内存中，因此我正在尝试解压缩每个块，处理数据并继续下一个块。

我正在使用以下代码：

public static String gzipDecompress(byte[] compressed) throws IOException {
    String uncompressed;
    try (
        ByteArrayInputStream bis = new ByteArrayInputStream(compressed);
        GZIPInputStream gis = new GZIPInputStream(bis);
        Reader reader = new InputStreamReader(gis);
        Writer writer = new StringWriter()
    ) {

      char[] buffer = new char[10240];
      for (int length = 0; (length = reader.read(buffer)) > 0; ) {
        writer.write(buffer, 0, length);
      }
      uncompressed = writer.toString();
    }

    return uncompressed;
  }

但是在使用第一个压缩块调用函数时出现以下异常：

java.io.EOFException: Unexpected end of ZLIB input stream
    at java.util.zip.InflaterInputStream.fill(InflaterInputStream.java:240)
    at java.util.zip.InflaterInputStream.read(InflaterInputStream.java:158)
    at java.util.zip.GZIPInputStream.read(GZIPInputStream.java:117)
    at sun.nio.cs.StreamDecoder.readBytes(StreamDecoder.java:284)
    at sun.nio.cs.StreamDecoder.implRead(StreamDecoder.java:326)
    at sun.nio.cs.StreamDecoder.read(StreamDecoder.java:178)
    at java.io.InputStreamReader.read(InputStreamReader.java:184)
    at java.io.Reader.read(Reader.java:140)

重要的是要提到我没有跳过任何块并尝试按顺序解压缩块。

我错过了什么？

Answer 1

问题是你手动玩这些块。

正确的方法是获取一些InputStream，用GZIPInputStream包装，然后读取数据。

    InputStream is = // obtain the original gzip stream

    GZIPInputStream gis = new GZIPInputStream(is);
    Reader reader = new InputStreamReader(gis);

    //... proceed reading and so on

GZIPInputStream 以流方式工作，因此如果您一次只从reader 请求 10kb，那么无论初始 GZIP 文件的大小如何，总体内存占用量都会很低。

问题更新后更新

针对您的情况，一个可能的解决方案是编写一个InputStream 实现，该实现将由您的客户端协议处理程序分块放入其中的字节流式传输。

这是一个原型：

public class ProtocolDataInputStream extends InputStream {
    private BlockingQueue<byte[]> nextChunks = new ArrayBlockingQueue<byte[]>(100);
    private byte[] currentChunk = null;
    private int currentChunkOffset = 0;
    private boolean noMoreChunks = false;

    @Override
    public synchronized int read() throws IOException {
        boolean takeNextChunk = currentChunk == null || currentChunkOffset >= currentChunk.length;
        if (takeNextChunk) {
            if (noMoreChunks) {
                // stream is exhausted
                return -1;
            } else {
                currentChunk = nextChunks.take();
                currentChunkOffset = 0;
            }
        }
        return currentChunk[currentChunkOffset++];
    }

    @Override
    public synchronized int available() throws IOException {
        if (currentChunk == null) {
            return 0;
        } else {
            return currentChunk.length - currentChunkOffset;
        }
    }

    public synchronized void addChunk(byte[] chunk, boolean chunkIsLast) {
        nextChunks.add(chunk);
        if (chunkIsLast) {
            noMoreChunks = true;
        }
    }
}

您的客户端协议处理程序使用addChunk() 添加字节块，而您的解压缩代码从该流中提取数据（通过Reader）。

请注意，此代码存在一些问题：

1. 正在使用的队列大小有限。如果addChunk()被调用太频繁，队列可能被填满，这将阻塞addChunk()。这可能是可取的，也可能不是。
2. 仅实现read() 方法以用于说明目的。为了性能，最好以同样的方式实现read(byte[])。
3. 在读取器（解压缩器）和写入器（调用 addChunk() 的协议处理程序）是不同线程的假设下使用保守同步。
4. InterruptedException 不在take() 上处理，以避免过多的细节。

如果你的解压器和addChunk()在同一个线程中执行（在同一个循环中），那么你可以尝试在使用InputStream或Reader.ready()拉时使用InputStream.available()方法在使用Reader拉时.

Answer 2

来自 gzip 流的任意字节序列不是有效的独立 gzip 数据。一种或另一种方式，您必须连接所有字节块。

最简单的方法是用一个简单的管道将它们全部累积起来：

import java.io.PipedOutputStream;
import java.io.IOException;
import java.util.zip.GZIPInputStream;

public class ChunkInflater {
    private final PipedOutputStream pipe;

    private final InputStream stream;

    public ChunkInflater()
    throws IOException {
        pipe = new PipedOutputStream();
        stream = new GZIPInputStream(new PipedInputStream(pipe));
    }

    public InputStream getInputStream() {
        return stream;
    }

    public void addChunk(byte[] compressedChunk)
    throws IOException {
        pipe.write(compressedChunk);
    }
}

现在您有了一个 InputStream，您可以按照您想要的任何增量进行读取。例如：

ChunkInflater inflater = new ChunkInflater();

Callable<Void> chunkReader = new Callable<Void>() {
    @Override
    public Void call()
    throws IOException {
        byte[] chunk;
        while ((chunk = readChunkFromSource()) != null) {
            inflater.addChunk(chunk);
        }

        return null;
    }
};
ExecutorService executor = Executors.newSingleThreadExecutor();
executor.submit(chunkReader);
executor.shutdown();

Reader reader = new InputStreamReader(inflater.getInputStream());
// read text here