计算 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 特征的文档：

如果一个 CONCURRENT 收集器也不是 UNORDERED，那么只有在应用于无序数据源时才应该同时评估它。

本文档的基本内容是，如果您的收集器是 CONCURRENT 并且流的源是 UNORDERED（如 Set）或者我们显式调用 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 }));

Answer 2

以下内容在保存摘要的本地类的帮助下使用可变归约。
收集元素的限制是通过在 combiner 函数中选择前 10 个元素来完成的。

使用IntStream的示例：

Stat result = IntStream.range(0, 100)
        .boxed()
        .filter(i -> i % 2 == 0)
        .collect(() -> new Stat(0, new ArrayList<Integer>()), 
            (stat, integer) -> {
                stat.count++;
                if (stat.list.size() < 10) {
                    stat.list.add(integer);
                }
            }, 
            (stat1, stat2) -> {
                stat1.list.addAll(stat2.list.subList(0, Math.min(stat2.list.size(), 
                    10 - stat1.list.size())));
            });

这是流中使用的Stat 类（您可以轻松使用Pair<Long, List<Integer>> 之类的东西）：

private static class Stat {
    long count;
    List<Integer> list;

    public Stat(long count, List<Integer> list) {
        this.count = count;
        this.list = list;
    }
}

以上示例结果为[count=50,list=[0, 2, 4, 6, 8, 10, 12, 14, 16, 18]]

Answer 3

这是一个简单的 lambda 表达式，它将计数添加到您的列表中的任何项目，最多 10 个，使其通过您的过滤器：

i -> {if (myList.size() < 10) myList.add(i);}

但你不能简单地使用count() on Stream：

如果一个实现能够直接从流源计算计数，则可以选择不执行流管道（顺序或并行）。在这种情况下，不会遍历源元素，也不会评估任何中间操作。有副作用的行为参数，除了调试等无害的情况外，强烈建议不要使用，可能会受到影响。

对于我的情况，使用count()，peek() 没有被调用，因为元素没有被遍历，我的列表是空的。

选择一个简单的归约来计算元素。

.reduce(0, (a, b) -> a + 1);

我的代码：

int count = yourCollection.stream()
    .filter(myFilter)
    .peek(i -> {if (myList.size() < 10) myList.add(i);} )
    .reduce(0, (a, b) -> a + 1);

Answer 4

这是另一种解决方案，不确定是否符合您的要求。

    final Count c = new Count();

    coll.stream().forEach(e -> {
        c.setTotCount(c.getTotCount() + 1);

        if (/*your filter*/) {
           // add till 10 elements only
           if (c.getMyList().size() <= 10) {
              c.addMyList(e);
           }
        }
    });

并且定义了辅助类

class Count {
    int totCount;
    // Student for an example
    List<Student> myList = new ArrayList<>();

    public List<Student> getMyList() {
        return myList;
    }

    public void addMyList(Student std) {
        this.myList.add(std);
    }

    // getter and setter for totCount
}

现在您有了列表和总数，它们都存储在帮助对象c 中。使用以下方法获取列表的总数：

  System.out.println(c.getTotCount());
  System.out.println(c.getMyList().size());