Java中可迭代的gzip压缩和deflate/inflate解压缩

Question

在互联网上是否有针对ByteBuffer进行gzip压缩的库？这种库可以让我们推入原始数据，然后拉取已经压缩的数据。我们已经搜索过了，但是只找到了处理InputStream和OutputStream的库。

我们的任务是在管道架构中为字节缓冲区创建gzip过滤器。这是一个拉模型体系结构，最后一个元素从前面的元素中拉取数据。我们的gzip过滤器处理ByteBuffers的流，没有单个的Stream对象可用。

我们曾试图将数据流适应成一种InputStream，然后使用GZipOutputStream来满足我们的需求，但是适配器代码量令人非常烦恼。

后接受编辑：记录一下，我们的架构类似于GStreamer等。

- Pedro Lamarão

---

我有点模糊地认为，要正确地gzip压缩整个输入，您必须一次性拥有全部输入 - 这可能是为什么它期望InputStream，而ByteBuffers通常用于存储中间数据，而不是保存整个文件的原因。 - Louis Wasserman

不是真正的gzip，但有JZlib，旨在提供“灵活”块中的（解）编码，与Java的ZIP支持函数相反。当然，Zlib不是gzip，但也许你仍然可以以某种方式利用它。 - JimmyB

@Hanno Java 7 包含了至关重要的 Deflater.SYNC_FLUSH 常量，可高效地传输压缩数据块。 - Simon G.

@Hanno，我们可以使用类似zlib的库中的基本元素来实现gzip，是的。如果我们找不到现成的解决方案，我们可能会向Github贡献一个。 - Pedro Lamarão

Answer 1

我不太理解“隐藏在互联网中”的部分，但是zlib提供内存中gzip格式的压缩和解压缩。虽然java.util.zip API提供了一些访问zlib的方法，但它受到限制。由于接口的限制，你不能直接请求zlib产生和消耗gzip流。但是，你可以使用nowrap选项来生成和消耗原始deflate数据。然后，你可以使用java.util.zip中的CRC32类轻松地制作自己的gzip头和尾。你可以在固定的10字节头之前添加，以小端顺序附加四字节CRC和四字节未压缩长度（对2的32次方取模），然后你就可以开始使用了。

- Mark Adler

---

“隐藏在互联网中”的部分指的是“我们已经搜索过了”，而事实上，我们找不到任何不需要可用的流对象的东西。” - Pedro Lamarão

哇，JZLib作者的回答。:-D - Simon G.

我是zlib的两位作者之一。JZlib是由Atsuhiko Yamanaka完成的将zlib翻译/适应到Java的工作。 - Mark Adler

啊，好的，我完全误读了那个 -- 我以为 ByteBuffers 在互联网中被隐藏了，因为它说“ByteBuffers hidden in the Internet”。 - Mark Adler

嗯，重新阅读我的问题，你的困惑是有道理的。这是我的错。 - Pedro Lamarão

Answer 2

非常感谢Mark Adler提供的这个方法，比我的原始答案要好得多。

package stack;

import java.io.*;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;
import java.util.zip.CRC32;
import java.util.zip.Deflater;

public class BufferDeflate2 {
    /** The standard 10 byte GZIP header */
    private static final byte[] GZIP_HEADER = new byte[] { 0x1f, (byte) 0x8b,
            Deflater.DEFLATED, 0, 0, 0, 0, 0, 0, 0 };

    /** CRC-32 of uncompressed data. */
    private final CRC32 crc = new CRC32();

    /** Deflater to deflate data */
    private final Deflater deflater = new Deflater(Deflater.BEST_COMPRESSION,
            true);

    /** Output buffer building area */
    private final ByteArrayOutputStream buffer = new ByteArrayOutputStream();

    /** Internal transfer space */
    private final byte[] transfer = new byte[1000];

    /** The flush mode to use at the end of each buffer */
    private final int flushMode;


    /**
     * New buffer deflater
     * 
     * @param syncFlush
     *            if true, all data in buffer can be immediately decompressed
     *            from output buffer
     */
    public BufferDeflate2(boolean syncFlush) {
        flushMode = syncFlush ? Deflater.SYNC_FLUSH : Deflater.NO_FLUSH;
        buffer.write(GZIP_HEADER, 0, GZIP_HEADER.length);
    }


    /**
     * Deflate the buffer
     * 
     * @param in
     *            the buffer to deflate
     * @return deflated representation of the buffer
     */
    public ByteBuffer deflate(ByteBuffer in) {
        // convert buffer to bytes
        byte[] inBytes;
        int off = in.position();
        int len = in.remaining();
        if( in.hasArray() ) {
            inBytes = in.array();
        } else {
            off = 0;
            inBytes = new byte[len];
            in.get(inBytes);
        }

        // update CRC and deflater
        crc.update(inBytes, off, len);
        deflater.setInput(inBytes, off, len);

        while( !deflater.needsInput() ) {
            int r = deflater.deflate(transfer, 0, transfer.length, flushMode);
            buffer.write(transfer, 0, r);
        }

        byte[] outBytes = buffer.toByteArray();
        buffer.reset();
        return ByteBuffer.wrap(outBytes);
    }


    /**
     * Write the final buffer. This writes any remaining compressed data and the GZIP trailer.
     * @return the final buffer
     */
    public ByteBuffer doFinal() {
        // finish deflating
        deflater.finish();

        // write all remaining data
        int r;
        do {
            r = deflater.deflate(transfer, 0, transfer.length,
                    Deflater.FULL_FLUSH);
            buffer.write(transfer, 0, r);
        } while( r == transfer.length );

        // write GZIP trailer
        writeInt((int) crc.getValue());
        writeInt((int) deflater.getBytesRead());

        // reset deflater
        deflater.reset();

        // final output
        byte[] outBytes = buffer.toByteArray();
        buffer.reset();
        return ByteBuffer.wrap(outBytes);
    }


    /**
     * Write a 32 bit value in little-endian order
     * 
     * @param v
     *            the value to write
     */
    private void writeInt(int v) {
        System.out.println("v="+v);
        buffer.write(v & 0xff);
        buffer.write((v >> 8) & 0xff);
        buffer.write((v >> 16) & 0xff);
        buffer.write((v >> 24) & 0xff);
    }


    /**
     * For testing. Pass in the name of a file to GZIP compress
     * @param args
     * @throws IOException
     */
    public static void main(String[] args) throws IOException {
        File inFile = new File(args[0]);
        File outFile = new File(args[0]+".test.gz");
        FileChannel inChan = (new FileInputStream(inFile)).getChannel();
        FileChannel outChan = (new FileOutputStream(outFile)).getChannel();

        BufferDeflate2 def = new BufferDeflate2(false);

        ByteBuffer buf = ByteBuffer.allocate(500);
        while( true ) {
            buf.clear();
            int r = inChan.read(buf);
            if( r==-1 ) break;
            buf.flip();
            ByteBuffer compBuf = def.deflate(buf);
            outChan.write(compBuf);
        }

        ByteBuffer compBuf = def.doFinal();
        outChan.write(compBuf);

        inChan.close();
        outChan.close();
    }
}

- Simon G.

---

这看起来像是我们首选的解决方案。 - Pedro Lamarão

嗯，看起来你这里有一个 bug；needsInput() 被检查是为了确定是否需要添加更多的输入，而 finished() 则是用于检查是否还有更多的输出，因此你应该将 needsInput() 的使用更改为 finished()。 - Jason S

Answer 3

处理ByteBuffer并不难。请看下面的示例代码。您需要知道如何创建缓冲区。选项如下：

1. 每个缓冲区都是独立压缩的。这很容易处理，我认为这不是问题。您只需将缓冲区转换为字节数组，并在GZIPInputStream中包装一个ByteArrayInputStream。
2. 每个缓冲区都以SYNC_FLUSH结束，因此包含流中的整个数据块。写入缓冲区的所有数据都可以立即被读取。
3. 每个缓冲区只是GZIP流的一部分。不能保证读者能从缓冲区读取任何内容。

GZIP生成的数据必须按顺序处理。ByteBuffers必须按照它们生成的顺序进行处理。

示例代码：

package stack;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.nio.ByteBuffer;
import java.nio.channels.Channels;
import java.nio.channels.Pipe;
import java.nio.channels.SelectableChannel;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.zip.GZIPInputStream;

public class BufferDeflate {

    static AtomicInteger idSrc = new AtomicInteger(1);

    /** Queue for transferring buffers */
    final BlockingQueue<ByteBuffer> buffers = new LinkedBlockingQueue<ByteBuffer>();

    /** The entry point for deflated buffers */
    final Pipe.SinkChannel bufSink;

    /** The source for the inflater */
    final Pipe.SourceChannel infSource;

    /** The destination for the inflater */
    final Pipe.SinkChannel infSink;

    /** The source for the outside world */
    public final SelectableChannel source;



    class Relayer extends Thread {
        public Relayer(int id) {
            super("BufferRelayer" + id);
        }


        public void run() {
            try {
                while( true ) {
                    ByteBuffer buf = buffers.take();
                    if( buf != null ) {
                        bufSink.write(buf);
                    } else {
                        bufSink.close();
                        break;
                    }
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }



    class Inflater extends Thread {
        public Inflater(int id) {
            super("BufferInflater" + id);
        }


        public void run() {
            try {
                InputStream in = Channels.newInputStream(infSource);
                GZIPInputStream gzip = new GZIPInputStream(in);
                OutputStream out = Channels.newOutputStream(infSink);

                int ch;
                while( (ch = gzip.read()) != -1 ) {
                    out.write(ch);
                }
                out.close();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }


    /**
     * New buffer inflater
     */
    public BufferDeflate() throws IOException {
        Pipe pipe = Pipe.open();
        bufSink = pipe.sink();
        infSource = pipe.source();

        pipe = Pipe.open();
        infSink = pipe.sink();
        source = pipe.source().configureBlocking(false);

        int id = idSrc.incrementAndGet();

        Thread thread = new Relayer(id);
        thread.setDaemon(true);
        thread.start();

        thread = new Inflater(id);
        thread.setDaemon(true);
        thread.start();
    }


    /**
     * Add the buffer to the stream. A null buffer closes the stream
     * 
     * @param buf
     *            the buffer to add
     * @throws IOException
     */
    public void add(ByteBuffer buf) throws IOException {
        buffers.offer(buf);
    }
}

只需将缓冲区传递给add方法，并从公共source通道中读取即可。在处理一定数量的字节后，可以从GZIP中读取的数据量是不可预测的。因此，我已将source通道设置为非阻塞，以便您可以在同一线程中安全地从中读取字节缓冲区。

- Simon G.

---

你的解决方案看起来是正确的，但需要太多资源。这是我们在尝试适应java.util.zip中的gzip流类时发现的。我们使用类似组件（例如加密过滤器）的经验告诉我们，这必须更简单、更便宜。 - Pedro Lamarão

太多的资源是为了什么？以上代码是在处理I/O阻塞时尽可能简单的实现。如果你想要便宜的解决方案，可以选择Mark Adler的答案。他是Java中处理zip格式文件的专家。 - Simon G.

对于这个问题来说，队列、管道和线程似乎有点过度了。 - Jason S

正如我之前所暗示的，如果你想要简单的话，就选择Mark Adler的答案吧 :) - Simon G.