java - 用 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)

在这种情况下，处理程序是实现具有此方法的接口(interface)的某个类:

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，这是您无法使用常规流实现的。