时间戳

首页 > 解决方案 > 由 Iterables.partition() 生成的拆分器没有按预期运行？

java - 由 Iterables.partition() 生成的拆分器没有按预期运行？

问题描述

我注意到使用番石榴产生的分裂Iterables.partition(collection, partitionSize).spliterator()器表现得很奇怪。

在结果拆分器上执行 trySplit() 不会拆分，但对初始 trySplit() 的结果执行 trySplit() 最终会拆分。

此外， usingStreamSupport.stream(Iterables.partition(collection, partitionSize).spliterator(), true)不会并行化流，而是 StreamSupport.stream(Iterables.partition(collection, partitionSize).spliterator().trySplit(), true)并行化并且生成的流包含所有分区。

我的目标是：给定一个大小为 100k 的集合，我想将它分成大小为 5000 的批次并并行处理这些批次。

2 个问题：Iterables.partition 生成的拆分器行为是否正确？我的方法是实现目标的好方法吗？

标签： javajava-streamguavapartitionspliterator

解决方案

这里的问题是Spliterator来自Iterable, 没有已知的大小。所以内部实现会将元素缓冲到一个大小1024的缓冲区中，并在下一次迭代中继续增加缓冲区。我的意思是：

    List<Integer> coll = IntStream.range(0, 150_000).boxed().collect(Collectors.toList());
    Iterable<List<Integer>> it = Iterables.partition(coll, 1);
    Spliterator<List<Integer>> sp = it.spliterator();

    Spliterator<List<Integer>> one = sp.trySplit();
    System.out.println(one.getExactSizeIfKnown());

    Spliterator<List<Integer>> two = sp.trySplit();
    System.out.println(two.getExactSizeIfKnown());

    Spliterator<List<Integer>> three = sp.trySplit();
    System.out.println(three.getExactSizeIfKnown());

    Spliterator<List<Integer>> four = sp.trySplit();
    System.out.println(four.getExactSizeIfKnown());

这将打印：

1024
2048
3072
4096

如果你想5000一次处理元素，你需要从Spliterator一个已知大小的开始。您可以将这些分区ArrayList放在首位：

 public static void main(String[] args) {

    List<Integer> coll = IntStream.range(0, 15_000).boxed().collect(Collectors.toList());
    Iterable<List<Integer>> it = Iterables.partition(coll, 5000);

    List<List<Integer>> list = new ArrayList<>();
    it.forEach(list::add);

    StreamSupport.stream(list.spliterator(), true)
            .map(x -> {
                System.out.println(
                        "Thread : " + Thread.currentThread().getName() +
                        " processed elements in the range : " + x.get(0) + " , " + x.get(x.size() - 1)
                );
                return x;
            })
            .flatMap(List::stream)
            .collect(Collectors.toList());
}

在我的机器上，它显示它们每个都由一个线程处理：

Thread : ForkJoinPool.commonPool-worker-5 processed elements in the range : 10000 , 14999
Thread : ForkJoinPool.commonPool-worker-19 processed elements in the range : 0 , 4999
Thread : main processed elements in the range : 5000 , 9999

推荐阅读