java - 计算 Stream 中的元素但只考虑 N 用于收集

Question

以下 lambda 在 Java 中是否可行？我想计算过滤后的流中的元素，但附带存储前 10 个元素

stream().filter(myFilter)  //Reduces input to forthcoming operations
        .limit(10)         //Limits to ten the amount of elements to finish stream 
        .peek(myList::add) //Stores the ten elements into a list
        .count();          //Here is the difficult one. Id like to count everything  the total of elements that pass the filter, beyond the 10 I am fetching

编辑:这对我来说太含蓄了，但这个想法当然意味着作为一个潜在的解决方案，它是最快的(比调用两次流生成器并在最少):

List<Entity> entities = stream().filter(myFilter) 
                                .limit(10)
                                .collect(Collectors.toList());
long entitiesCount = stream().filter(myFilter) 
                             .count();

...从单次迭代中获益，而无需将整个集合加载到内存中。我正在对答案进行并行化测试

Answer 1

自定义收集器就是这里的答案:

Entry<List<Integer>, Integer> result = list.stream()
            .collect(Collector.of(
                    () -> new SimpleEntry<>(new ArrayList<>(), 0),
                    (l, x) -> {
                        if (l.getKey().size() < 10) {
                            l.getKey().add(x);
                        }
                        l.setValue(l.getValue() + 1);
                    },
                    (left, right) -> {
                        List<Integer> leftList = left.getKey();
                        List<Integer> rightList = right.getKey();
                        while (leftList.size() < 10 && rightList.size() > 0) {
                            leftList.add(rightList.remove(0));
                        }
                        left.setValue(left.getValue() + right.getValue());
                        return left;
                    }));

假设你有这样的代码:

Set.of(1, 2, 3, 4)
            .stream()
            .parallel()
            .collect(Collector.of(
                    ArrayList::new,
                    (list, ele) -> {
                        System.out.println("Called accumulator");
                        list.add(ele);
                    },
                    (left, right) -> {
                        System.out.println("Combiner called");
                        left.addAll(right);
                        return left;
                    },
                    new Characteristics[] { Characteristics.CONCURRENT }));

在我们开始考虑该代码之前(对于示例的目的来说，它的正确性很重要)，我们需要稍微阅读文档以了解 CONCURRENT 特性:

If a CONCURRENT collector is not also UNORDERED, then it should only be evaluated concurrently if applied to an unordered data source.

这个文档基本上说的是，如果你的收集器是CONCURRENT 并且流的源是UNORDERED(就像一个设置 ) 或者我们显式调用 unordered 那么合并将永远不会被调用。

如果您运行前面的代码，您将看到 Combiner called 永远不会出现在输出中。

如果将 Set.of(1, 2, 3, 4) 更改为 List.of(1, 2, 3, 4)，您将看到不同的图片(忽略你得到的结果的正确性 - 因为 ArrayList 不是线程安全的，但这不是重点)。如果流的源是 List 并且同时调用 unordered 你会再次看到只有累加器是称为，即:

 List.of(1, 2, 3, 4)
            .stream()
            .unordered()
            .parallel()
            .collect(Collector.of(
                    ArrayList::new,
                    (list, ele) -> {
                        System.out.println("Called accumulator");
                        list.add(ele);
                    },
                    (left, right) -> {
                        System.out.println("Combiner called");
                        left.addAll(right);
                        return left;
                    },
                    new Characteristics[] { Characteristics.CONCURRENT }));