对Java 8流进行分区

Question

如何在Java8 Stream上实现“分区”操作？通过分区，我的意思是把一个流分成给定大小的子流。不知何故，它将与GuavaIterators.partition（）方法相同，只是希望分区是延迟计算的Streams而不是List的。

Answer 1

如果您想按顺序使用Stream，就可以对Stream进行分区（以及执行相关功能，例如窗口化——我认为这是您在这种情况下真正想要的）。支持标准Streams分区的两个库是独眼巨人-反应（我是作者）和jOλ，独眼巨人-反应扩展了它（添加窗口化等功能）。

cyclops-streams有一个静态函数StreamUtils的集合，用于对Java流进行操作，还有一系列函数如splitAt、headAndTail、splitBy、partition用于分区。

要将流窗口化为大小为30的嵌套流，可以使用window方法。

对于OPs来说，用流的术语来说，将一个流分成多个给定大小的流是一种窗口操作(而不是分区操作)。

  Stream<Streamable<Integer>> streamOfStreams = StreamUtils.window(stream,30);

有一个叫做ReactiveSeq的流扩展类，它扩展了jool。Seq，并添加了窗口功能，这可能会使代码更加简洁。

  ReactiveSeq<Integer> seq;
  ReactiveSeq<ListX<Integer>> streamOfLists = seq.grouped(30);

正如 Tagir 上面指出的那样，这不适合并行流。如果要对希望以多线程方式执行的流进行窗口或批处理。独眼巨人反应中的LazyFutureStream可能很有用（窗口在待办事项列表中，但现在可以使用普通的旧批处理）。

在这种情况下，数据将从执行流的多个线程传递到多生产者/单消费者无等待队列，并且来自该队列的顺序数据可以在被再次分发到线程之前被窗口化。

  Stream<List<Data>> batched = new LazyReact().range(0,1000)
                                              .grouped(30)
                                              .map(this::process);

Answer 2

我找到了一个优雅的解决方案：可迭代部分=Iterables.partition（流::迭代器，大小）

Answer 3

不可能将任意的源流划分为固定大小的批次，因为这将破坏并行处理。并行处理时，您可能不知道拆分后的第一个子任务中有多少元素，因此在第一个子任务完全处理完之前，您无法为下一个子任务创建分区。

但是，可以从随机访问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);

如果不想依赖第三方库，可以手动实现这样的< code > of publists 方法:

public static <T> Stream<List<T>> ofSubLists(List<T> source, int length) {
    if (length <= 0)
        throw new IllegalArgumentException("length = " + length);
    int size = source.size();
    if (size <= 0)
        return Stream.empty();
    int fullChunks = (size - 1) / length;
    return IntStream.range(0, fullChunks + 1).mapToObj(
        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, 
                   Collector<List<T>, A, R> downstream) {
    class Acc {
        List<T> cur = new ArrayList<>();
        A acc = downstream.supplier().get();
    }
    BiConsumer<Acc, T> accumulator = (acc, t) -> {
        acc.cur.add(t);
        if(acc.cur.size() == batchSize) {
            downstream.accumulator().accept(acc.acc, acc.cur);
            acc.cur = new ArrayList<>();
        }
    };
    return Collector.of(Acc::new, accumulator,
            (acc1, acc2) -> {
                acc1.acc = downstream.combiner().apply(acc1.acc, acc2.acc);
                for(T t : acc2.cur) accumulator.accept(acc1, t);
                return acc1;
            }, acc -> {
                if(!acc.cur.isEmpty())
                    downstream.accumulator().accept(acc.acc, acc.cur);
                return downstream.finisher().apply(acc.acc);
            }, Collector.Characteristics.UNORDERED);
}

用法示例：

List<List<Integer>> list = IntStream.range(0,20)
                                    .boxed().parallel()
                                    .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,
        Collector<T, AA, B> batchCollector,
        Collector<B, A, R> downstream) {
    return unorderedBatches(batchSize, 
            Collectors.mapping(list -> list.stream().collect(batchCollector), downstream));
}

例如，通过这种方式，您可以即时对每批数据求和:

List<Integer> list = IntStream.range(0,20)
        .boxed().parallel()
        .collect(unorderedBatches(3, Collectors.summingInt(Integer::intValue), 
            Collectors.toList()));