将Stream<Object>转换为InputStream

Question

我该如何将类型为Stream<Object>的数据转换为InputStream？目前，我会获取迭代器并循环遍历所有数据，将其转换为字节数组并添加到InputStream中：

 ByteArrayOutputStream bos = new ByteArrayOutputStream();
 ObjectOutputStream oos = new ObjectOutputStream(bos);

 Iterator<MyType> myItr = MyObject.getStream().iterator();

 while (myItr.hasNext()) {   

       oos.writeObject(myItr.next().toString()
         .getBytes(StandardCharsets.UTF_8));
   }
   oos.flush();
   oos.close();

   InputStream is = new ByteArrayInputStream(bao.toByteArray());

但是这样做的开销会有多大呢？如果我的流包含一太字节的数据，那么我不就要将一太字节的数据吸入内存中吗？有没有更好的方法来实现这个目标呢？

- BigBug

---

你确定这种类型的InputStream是你需要的吗？你正在将对象转换为字符串，获取它们的UTF-8表示作为字节数组，并对这些数组对象使用对象序列化。目前不清楚你想在另一端接收什么，当前既不是对象也不是字符串。你可以直接编写字符串，或者可以使用Writer编写一个更简单的纯文本表示，而不需要对象序列化协议的开销，但两者都无法重新创建原始对象。 - Holger

Answer 1

这个对您有用吗？

这是一个InputStream实现，它将Stream作为输入数据。 您只需要.map()您的任意对象到字节数组，以任何您想要的方式表示每个对象为字节。

当读取InputStream时，它仅在Stream上调用终端操作，并在使用者读取更多InputStream时从Stream中获取对象，因此它永远不会将整个集合加载到内存中。 https://gist.github.com/stephenhand/292cdd8bba7a452d83c51c00d9ef113c

- Stephen James Hand

Answer 2

您应该能够使用管道将 OutputStream 转换为 InputStream ：

PipedOutputStream pos = new PipedOutputStream();
InputStream is = new PipedInputStream(pos);

new Thread(() -> {
    try (ObjectOutputStream oos = new ObjectOutputStream(pos)) {
        Iterator<MyType> myItr = MyObject.getStream().iterator();
        while (myItr.hasNext()) {
            oos.writeObject(myItr.next().toString()
                .getBytes(StandardCharsets.UTF_8));
        }
    } catch (IOException e) {
        // handle closed pipe etc.
    }
}).start();

受这个答案的启发。

- shmosel

---

嗯...我不太明白这段代码的作用是什么？pos从未被分配任何数据 :S，看起来我们正在循环遍历并将myItr的内容添加到oos中，但没有任何理由将其与输入流包含的内容联系起来？我有什么地方理解错了吗？ - BigBug

“is” 和 “oos” 都与 “pos” 相关联，它将数据从输出流传输到输入流。但是现在我想起来了，这仍然需要在内存中缓冲所有数据，所以这并没有真正解决你的问题。 - shmosel

@BigBug 好的，我之前关于它将所有内容缓存到内存中的说法是错误的。实际上，一旦达到缓冲区大小，它会阻塞，这意味着你必须在单独的线程上提供输出流。请查看我的更新答案。 - shmosel