分区Java 8流

Question

如何在Java 8 Stream上实现“分区”操作？通过分区，我的意思是将一个流分成给定大小的子流。它在某种程度上与Guava Iterators.partition（）方法相同，只是希望分区是延迟评估的Streams而不是List的。

Answer 1

将任意源流划分为固定大小的批次是不可能的，因为这会加重并行处理。并行处理时，您可能不知道拆分后的第一个子任务中有多少个元素，因此您无法为下一个子任务创建分区，直到完全处理第一个子任务。但是，可以从随机访问中创建分区流List。此类功能在例如我的StreamEx图书馆中可用：List<Type> input = Arrays.asList(...);Stream<List<Type>> stream = StreamEx.ofSubLists(input, partitionSize);或者，如果您真的想要流：Stream<Stream<Type>> stream = StreamEx.ofSubLists(input, partitionSize).map(List::stream);如果您不想依赖第三方库，则可以ofSubLists手动实现这种方法：public static <T> Stream<List<T>> ofSubLists(List<T> source, int length) {&nbsp; &nbsp; if (length <= 0)&nbsp; &nbsp; &nbsp; &nbsp; throw new IllegalArgumentException("length = " + length);&nbsp; &nbsp; int size = source.size();&nbsp; &nbsp; if (size <= 0)&nbsp; &nbsp; &nbsp; &nbsp; return Stream.empty();&nbsp; &nbsp; int fullChunks = (size - 1) / length;&nbsp; &nbsp; return IntStream.range(0, fullChunks + 1).mapToObj(&nbsp; &nbsp; &nbsp; &nbsp; n -> source.subList(n * length, n == fullChunks ? size : (n + 1) * length));}该实现看起来有些长，但是考虑了一些极端情况，例如接近MAX_VALUE列表大小。如果您想为无序流提供并行友好的解决方案（因此您不必担心将哪些流元素合并为单个批处理），则可以使用这样的收集器（感谢@sibnick的启发）：public static <T, A, R> Collector<T, ?, R> unorderedBatches(int batchSize,&nbsp;&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;Collector<List<T>, A, R> downstream) {&nbsp; &nbsp; class Acc {&nbsp; &nbsp; &nbsp; &nbsp; List<T> cur = new ArrayList<>();&nbsp; &nbsp; &nbsp; &nbsp; A acc = downstream.supplier().get();&nbsp; &nbsp; }&nbsp; &nbsp; BiConsumer<Acc, T> accumulator = (acc, t) -> {&nbsp; &nbsp; &nbsp; &nbsp; acc.cur.add(t);&nbsp; &nbsp; &nbsp; &nbsp; if(acc.cur.size() == batchSize) {&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; downstream.accumulator().accept(acc.acc, acc.cur);&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; acc.cur = new ArrayList<>();&nbsp; &nbsp; &nbsp; &nbsp; }&nbsp; &nbsp; };&nbsp; &nbsp; return Collector.of(Acc::new, accumulator,&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; (acc1, acc2) -> {&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; acc1.acc = downstream.combiner().apply(acc1.acc, acc2.acc);&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; for(T t : acc2.cur) accumulator.accept(acc1, t);&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; return acc1;&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; }, acc -> {&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; if(!acc.cur.isEmpty())&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; downstream.accumulator().accept(acc.acc, acc.cur);&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; return downstream.finisher().apply(acc.acc);&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; }, Collector.Characteristics.UNORDERED);}用法示例：List<List<Integer>> list = IntStream.range(0,20)&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; .boxed().parallel()&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; .collect(unorderedBatches(3, Collectors.toList()));结果：[[2, 3, 4], [7, 8, 9], [0, 1, 5], [12, 13, 14], [17, 18, 19], [10, 11, 15], [6, 16]]这样的收集器是完全线程安全的，并为顺序流生成有序批。如果要对每个批次应用中间转换，则可以使用以下版本：public static <T, AA, A, B, R> Collector<T, ?, R> unorderedBatches(int batchSize,&nbsp; &nbsp; &nbsp; &nbsp; Collector<T, AA, B> batchCollector,&nbsp; &nbsp; &nbsp; &nbsp; Collector<B, A, R> downstream) {&nbsp; &nbsp; return unorderedBatches(batchSize,&nbsp;&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; Collectors.mapping(list -> list.stream().collect(batchCollector), downstream));}例如，通过这种方式，您可以即时对每个批次中的数字求和：List<Integer> list = IntStream.range(0,20)&nbsp; &nbsp; &nbsp; &nbsp; .boxed().parallel()&nbsp; &nbsp; &nbsp; &nbsp; .collect(unorderedBatches(3, Collectors.summingInt(Integer::intValue),&nbsp;&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; Collectors.toList()));

Answer 2

正如Jon Skeet在其评论中所示，似乎不可能使分区变得懒惰。对于非延迟分区，我已经有了以下代码：public static <T> Stream<Stream<T>> partition(Stream<T> source, int size) {&nbsp; &nbsp; final Iterator<T> it = source.iterator();&nbsp; &nbsp; final Iterator<Stream<T>> partIt = Iterators.transform(Iterators.partition(it, size), List::stream);&nbsp; &nbsp; final Iterable<Stream<T>> iterable = () -> partIt;&nbsp; &nbsp; return StreamSupport.stream(iterable.spliterator(), false);}