Java的GZIPOutputStream拒绝刷新

Question

下面是一个JUnit测试，它演示了我的问题：

package stream;

import static org.junit.jupiter.api.Assertions.*;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.io.OutputStream;
import java.io.OutputStreamWriter;
import java.io.PrintWriter;
import java.nio.ByteBuffer;
import java.util.Arrays;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.zip.GZIPInputStream;
import java.util.zip.GZIPOutputStream;

import org.junit.jupiter.api.Test;

class StreamTest {
    public static class LoopbackStream {
        private final byte[] END_MARKER = new byte[0];
        private final ArrayBlockingQueue<byte[]> queue = new ArrayBlockingQueue<>(1024);

        public OutputStream getOutputStream() {
            return new OutputStream() {
                @Override
                public void write(int b) throws IOException {
                    this.write(new byte[] { (byte) b });
                }

                @Override
                public void write(byte[] b, int off, int len) {
                    try {
                        queue.put(Arrays.copyOfRange(b, off, len - off));
                    } catch (InterruptedException e) {
                        Thread.currentThread().interrupt();
                    }
                }

                @Override
                public void close() {
                    try {
                        queue.put(END_MARKER);
                    } catch (InterruptedException e) {
                        Thread.currentThread().interrupt();
                    }
                }
            };
        }

        public InputStream getInputStream() {
            return new InputStream() {
                private boolean finished = false;
                private ByteBuffer current = ByteBuffer.wrap(new byte[0]);

                @Override
                public int read() {
                    if (ensureData()) {
                        return Byte.toUnsignedInt(current.get());
                    } else {
                        return -1;
                    }
                }

                @Override
                public int read(byte[] b, int off, int len) {
                    if (ensureData()) {
                        int position = current.position();
                        current.get(b, off, Math.min(len, current.remaining()));
                        return current.position() - position;
                    } else {
                        return -1;
                    }
                }

                private boolean ensureData() {
                    if (!finished && !current.hasRemaining()) {
                        try {
                            byte[] data = queue.take();
                            current = ByteBuffer.wrap(data);
                            finished = data == END_MARKER;
                        } catch (InterruptedException e) {
                            Thread.currentThread().interrupt();
                            return false;
                        }
                    }
                    return !finished;
                }
            };
        }

    }

    @Test
    void testVanilla() throws IOException {
        LoopbackStream objectUnderTest = new LoopbackStream();

        PrintWriter pw = new PrintWriter(new OutputStreamWriter(objectUnderTest.getOutputStream()), true);
        BufferedReader br = new BufferedReader(new InputStreamReader(objectUnderTest.getInputStream()));

        pw.println("Hello World!");

        assertEquals("Hello World!", br.readLine());
    }

    @Test
    void testVanilla2() throws IOException {
        LoopbackStream objectUnderTest = new LoopbackStream();

        PrintWriter pw = new PrintWriter(new OutputStreamWriter(objectUnderTest.getOutputStream()), true);
        BufferedReader br = new BufferedReader(new InputStreamReader(objectUnderTest.getInputStream()));

        pw.println("Hello World!");
        assertEquals("Hello World!", br.readLine());

        pw.println("Hello Otherworld!");
        assertEquals("Hello Otherworld!", br.readLine());
    }

    @Test
    void testGzipped() throws IOException {
        LoopbackStream objectUnderTest = new LoopbackStream();

        PrintWriter pw = new PrintWriter(new OutputStreamWriter(new GZIPOutputStream(objectUnderTest.getOutputStream(), true)), true);
        BufferedReader br = new BufferedReader(new InputStreamReader(new GZIPInputStream(objectUnderTest.getInputStream())));

        pw.println("Hello World!");

        assertEquals("Hello World!", br.readLine());
    }
}

有两个单独的测试。一个使用普通的输入流和输出流(它工作得很好)，另一个用gzip等效项包装这些流。

我使用了GZIPOutputStream的syncFlush选项，每当刷新父流时，我希望它自动刷新流中的任何剩余字节。我使用PrintWriter的autoFlush选项，每当它执行println时，刷新它的数据。

有没有更好的方法来强制GZIPOutputStream在println之后刷新它的缓冲区

Answer 1

我知道这不是你问题的完整答案，但评论太长了……

更新：

进一步研究后发现，不刷新的似乎不是GZIPOutputStream (通过在public void write(byte[] b, int off, int len)方法中添加System.out.println("xy");语句，您可以看到GZIPOutputStream将两个字节数组写入OutputStream：一个是gzip流头，另一个是第一行文本的编码内容)。

读取过程似乎阻塞了，因为java.io.InputStreamReader (分别是它使用的sun.nio.cs.StreamDecoder )和GZIPInputStream之间的交互不好。

基本上，如果StreamDecoder需要从底层流读取字节，它会尝试读取尽可能多的字节(只要底层流报告in.available() > 0，这意味着底层流可以在不阻塞的情况下产生更多字节)

• loop in the StreamDecoder
• check for in.available()

此检查的问题在于，InflaterInputStream (GZIPInputStream的超类)始终返回可用字节数的1，即使其源流没有可用的字节(see the source of InflaterInputStream.available())也是如此

因此，虽然您可以逐行将数据写入GZIPOutputStream，但从GZIPInputStream中逐行读取数据似乎并不容易。

原始答案：

问题不在于GZIPOutputStream，而在于拒绝读取多个块的boolean ensureData()方法。

下面的测试对于vanilla流也会失败：

@Test
void testVanilla2() throws IOException {
    LoopbackStream objectUnderTest = new LoopbackStream();

    PrintWriter pw = new PrintWriter(new OutputStreamWriter(objectUnderTest.getOutputStream()), true);
    BufferedReader br = new BufferedReader(new InputStreamReader(objectUnderTest.getInputStream()));

    pw.println("Hello World!");
    assertEquals("Hello World!", br.readLine());

    pw.println("Hello Otherworld!");
    assertEquals("Hello Otherworld!", br.readLine());
}