输入流输出流是以内存为标准_构建用于测试的超大内存输入流

输入流输出流是以内存为标准

由于某种原因，我需要非常大的，甚至可能是无限的InputStream ，它会反复地反复返回相同的byte[] 。这样，我可以通过重复小样本来产生疯狂的大数据流。可以在Guava中找到类似的功能： Iterable<T> Iterables.cycle(Iterable<T>)和Iterator<T> Iterators.cycle(Iterator<T>) 。例如，如果您需要0和1的无限来源，只需说出Iterables.cycle(0, 1)并无限获取Iterables.cycle(0, 1) 0, 1, 0, 1, 0, 1... 不幸的是，我还没有为InputStream找到这样的工具，所以我跳入自己编写的工具。本文记录了我在此过程中犯的许多错误，主要是由于过于复杂和过度设计的简单解决方案。

我们实际上并不需要无限的InputStream ，能够创建非常大的InputStream （例如32 GiB）就足够了。因此，我们采用以下方法：

public static InputStream repeat(byte[] sample, int times)

它基本上采用字节sample数组，并返回一个InputStream返回这些字节。但是，当sample用完时，它将翻转，再次返回相同的字节-重复此过程指定的次数，直到InputStream信号结束。我尚未真正尝试过但似乎最显而易见的一种解决方案：

public static InputStream repeat(byte[] sample, int times) {final byte[] allBytes = new byte[sample.length * times];for (int i = 0; i < times; i++) {System.arraycopy(sample, 0, allBytes, i * sample.length, sample.length);}return new ByteArrayInputStream(allBytes);
}

我看到你在那笑！如果sample是100个字节，并且我们需要32 GiB的输入重复这100个字节，则生成的InputStream不应真正分配32 GiB的内存，我们在这里必须更加聪明。实际上，上面的repeat()有另一个细微的错误。 Java中的数组限制为2 ³¹ -1个条目（ int ），比其高32 GiB。该程序编译的原因是此处无声的整数溢出： sample.length * times 。此乘法不适用于int 。

好的，让我们尝试至少在理论上可行的方法。我的第一个想法是：如果我创建多个ByteArrayInputStream共享相同的byte[] sample （它们不进行急切的复制），然后以某种方式将它们连接在一起怎么办？因此，我需要一些InputStream适配器，该适配器可以采用任意数量的基础InputStream并将它们链接在一起-当第一个流耗尽时，切换到下一个。当您在Apache Commons或Guava中寻找某些东西，并且显然永远在JDK中时，这一尴尬时刻… java.io.SequenceInputStream几乎是理想的选择。但是，它只能精确地链接两个基础InputStream 。当然，由于SequenceInputStream本身就是InputStream ，我们可以递归地将其用作外部SequenceInputStream的参数。重复此过程，我们可以将任意数量的ByteArrayInputStream在一起：

public static InputStream repeat(byte[] sample, int times) {if (times <= 1) {return new ByteArrayInputStream(sample);} else {return new SequenceInputStream(new ByteArrayInputStream(sample),repeat(sample, times - 1));}
}

如果times为1，则将sample包装在ByteArrayInputStream 。否则，递归使用SequenceInputStream 。我认为您可以立即发现此代码的问题：太深的递归。嵌套级别与times参数相同，将达到数百万甚至数十亿。肯定有更好的办法。幸运的是，较小的改进将递归深度从O（n）更改为O（logn）：

public static InputStream repeat(byte[] sample, int times) {if (times <= 1) {return new ByteArrayInputStream(sample);} else {return new SequenceInputStream(repeat(sample, times / 2),repeat(sample, times - times / 2));}
}

老实说，这是我尝试的第一个实现。这是分而治之原理的简单应用，在这里我们将结果平均分成两个较小的子问题。看起来很聪明，但是有一个问题：容易证明我们创建了t（ t =times ） ByteArrayInputStreams和O（t） SequenceInputStream 。共享sample字节数组时，数百万个各种InputStream实例浪费了内存。这导致我们可以选择实现，无论times多少， InputStream创建一个InputStream ：

import com.google.common.collect.Iterators;
import org.apache.commons.lang3.ArrayUtils;public static InputStream repeat(byte[] sample, int times) {final Byte[] objArray = ArrayUtils.toObject(sample);final Iterator<Byte> infinite = Iterators.cycle(objArray);final Iterator<Byte> limited = Iterators.limit(infinite, sample.length * times);return new InputStream() {@Overridepublic int read() throws IOException {return limited.hasNext() ?limited.next() & 0xFF :-1;}};
}

毕竟，我们将使用Iterators.cycle() 。但是在我们必须将byte[]转换为Byte[]因为迭代器只能与objets一起使用，而不能与原语一起使用。没有惯用的方法ArrayUtils.toObject(byte[])语数组转换为盒装类型数组，因此我使用来自Apache Commons Lang的ArrayUtils.toObject(byte[]) 。有了对象数组，我们可以创建一个infinite迭代器，循环遍历sample值。由于我们不需要无限的流，因此再次使用了来自Guava的Iterators.limit(Iterator<T>, int)来切断无限迭代Iterators.limit(Iterator<T>, int) 。现在，我们只需要从Iterator<Byte>到InputStream进行桥接–毕竟它们在语义上代表同一件事。

该解决方案遭受两个问题。首先，由于拆箱，它会产生大量垃圾。垃圾回收并不太关心死的，短命的物品，但看起来仍然很浪费。我们之前已经遇到的第二个问题： sample.length * times乘以倍数可能会导致整数溢出。由于Iterators.limit()占用int不long ，因此没有固定的理由，因此无法修复。顺便说一句，我们通过按位和 0xFF避免了第三个问题-否则，值为-1 byte将表示流的结束，事实并非如此。 x & 0xFF已正确转换为无符号255 （ int ）。

因此，即使上面的实现是简短而甜美，声明式而不是命令式的，但它仍然太慢且受限制。如果您有C背景，我可以想象您看到我挣扎时会感到多么不舒服。在我最后想到的是最简单，痛苦的简单和低级的实现之后：

public static InputStream repeat(byte[] sample, int times) {return new InputStream() {private long pos = 0;private final long total = (long)sample.length * times;public int read() throws IOException {return pos < total ?sample[(int)(pos++ % sample.length)] :-1;}};
}

无GC的纯JDK，快速，简单易懂。让这成为您的一个教训：从您想到的最简单的解决方案开始，不要过度设计并且不要太聪明。我以前的解决方案（声明性，功能性，不变性等）–也许它们看起来很聪明，但是它们既不快速也不容易理解。

我们刚刚开发的实用程序不仅是一个玩具项目，还将在后续文章中使用。