用 BufferedInputStream 包装 PipedInputStream

Question

我有一个需要读取的 OutputStream，因此我使用以下（Groovy）代码来获取对数据的 InputStream 引用：

PipedInputStream inputStream = new PipedInputStream()
    PipedOutputStream outputStream = new PipedOutputStream(inputStream)
    new Thread(
            new Runnable() {
                public void run() {
                    // Some API method
                    putDataInOutputStream(outputStream)
                    outputStream.close()
                }
            }
    ).start()

handler.process(inputStream)

这种情况下，处理程序是一些实现具有此方法的接口的类：

public void process(InputStream stream);

在 我们的新要求是对流进行一些预处理，因此我需要在 handler.process() 方法中至少读取流两次。以下是一种实现的一些示例代码：

public void process(InputStream stream) {
    def bufferedStream = new BufferedInputStream(stream, 30 * 1048576)  // 30 MB
    bufferedStream.mark(Integer.MAX_VALUE)
    parseMetadata(bufferedStream)
    bufferedStream.reset()
    doTheThingYouDo(bufferedStream)
 }

我知道对于某些输入，我未达到 30 MB 限制或 Integer.MAX_VALUE 缓冲区大小。但是，我总是遇到以下异常：

java.io.IOException: Stream closed

这可能吗？我认为问题在于 PipedOutputStream 上的线程关闭，但我不知道如何防止这种情况，或者我作为 Java Stream IO 的新手是否会造成更多问题。

Answer 1

我最好的猜测是您的 parseMetadata 以某种方式关闭了流。我已经尝试过你的方案，它对我来说效果很好。一般来说，在处理程序完成读取之前关闭 OutputStream 不是问题，这正是管道流的用途。

此外，考虑到您的情况，我会省略管道和附加螺纹。如果您不介意将整个流都放在内存中，则可以执行类似的操作，

ByteArrayOutputStream out = new ByteArrayOutputStream();

fillTheOutput(out);

ByteArrayInputStream in = new ByteArrayInputStream(out.toByteArray());

pass1(in);
in.reset();
pass2(in);

如果您确实介意将所有内容都放在内存中，那么无论如何您都会遇到麻烦，因为您的 BufferedInputStream 执行的操作大致相同。

编辑：请注意，您可以轻松地基于字节数组构建新的ByteArrayInputStream，这是常规流无法做到的。