从 InputStream 创建 java.util.stream.Stream 的最佳方法是什么？答案

【问题标题】：What is the best way to create a java.util.stream.Stream from an InputStream?从 InputStream 创建 java.util.stream.Stream 的最佳方法是什么？
【发布时间】：2020-03-03 18:32:59
【问题描述】：

据我了解，InputStream 是一个字节流。我有兴趣将 InputStream 对象转换为字节流。基本上是以下方法的实现。

public Stream<byte[]> toStream(final InputStream is, final int bufferSize);

完成这项工作的最佳方法是什么？缓冲区大小是一次从 InputStream 读取的字节数。

【问题讨论】：

您目前如何阅读/处理您的InputStream？
为什么要Stream<byte[]>？你打算如何处理它？
@thisisshantzz 那么您正在寻找错误的方法。您不需要Stream，您需要将InputStream 连接到客户端的OutputStream，然后正常流式传输字节。具体如何完成取决于您的环境，但例如对于 Spring，您可以使用 StreamingResponseBody。
在您发布的方法的签名中，您试图将 InputStream 转换为字节数组流而不是字节流。首先，您必须解释该字节数组的含义。如果它们是文本行，您可以使用此 stackoverflow.com/questions/30336257/...。如果你想要字节流而不是字节数组流，你可以使用 inputStream 的 readAllBytes 方法，然后将数组转换为 Stream
@thisisshantzz 如果您只想将InputStream“重定向”到客户端，那么在两者之间任何额外的Flux 或byte[] 都是完全没有必要的。这只是普通的基本流媒体。请参阅here 了解更多信息。

标签： java java-stream inputstream

【解决方案1】：

你需要自己写Spliterator，类似这样：

public final class ChunkingInputStreamSpliterator implements Spliterator<byte[]> {

    private final InputStream is;
    private final int bufferSize;

    public ChunkingInputStreamSpliterator(InputStream is, int bufferSize) {
        this.is = is;
        this.bufferSize = bufferSize;
    }

    @Override
    public boolean tryAdvance(Consumer<? super byte[]> action) {
        byte[] bytes;
        try {
            bytes = this.is.readNBytes(this.bufferSize);
        } catch (IOException e) {
            throw new UncheckedIOException(e);
        }
        if (bytes.length == 0)
            return false;
        action.accept(bytes);
        return true;
    }

    @Override
    public Spliterator<byte[]> trySplit() {
        return null; // cannot split an InputStream
    }

    @Override
    public long estimateSize() {
        return Long.MAX_VALUE; // unknown
    }

    @Override
    public int characteristics() {
        return Spliterator.ORDERED | Spliterator.NONNULL;
    }

}

然后像这样实现你的方法：

public static Stream<byte[]> toStream(InputStream is, int bufferSize) {
    return StreamSupport.stream(new ChunkingInputStreamSpliterator(is, bufferSize), false);
}

如果你没有 Java 11，所以你没有非常方便的 readNBytes 方法，那么你自己做这部分是这样的：

public boolean tryAdvance(Consumer<? super byte[]> action) {
    byte[] bytes = new byte[this.bufferSize];
    int len = 0;
    try {
        for (int read; len < bytes.length; len += read)
            if ((read = this.is.read(bytes, len, bytes.length - len)) <= 0)
                break;
    } catch (IOException e) {
        throw new UncheckedIOException(e);
    }
    if (len == 0)
        return false;
    if (len < bytes.length)
        bytes = Arrays.copyOfRange(bytes, 0, len);
    action.accept(bytes);
    return true;
}

【讨论】：

在 Spliterator 的 estimatedSize 方法中将 Long.MAX_VALUE 替换为 this.is.available()/this.bufferSize 是否明智？
@thisissantzz 如果您知道尺寸，例如因为它是一个文件，所以您应该预先提供大小（尽管在这种情况下，FileChannel 会更有效）。否则，InputStream.available() 不会提供正确的大小。它只告诉没有阻塞可以读取多少。