java - Java 8 Streams 中的可变参数

Question

看这个问题:How to dynamically do filtering in Java 8?

问题是在执行过滤器后截断流。我不能使用 limit因为我不知道列表在过滤器之后有多长。那么，我们可以计算过滤后的元素吗？

所以，我想我可以创建一个类来计数并通过 map 传递流。 The code is in this answer .

我创建了一个计数但保留元素不变的类，我在这里使用一个函数，以避免使用我在其他答案中使用的 lambda:

class DoNothingButCount<T > implements Function<T, T> {
    AtomicInteger i;
    public DoNothingButCount() {
        i = new AtomicInteger(0);
    }
    public T apply(T p) {
        i.incrementAndGet();
        return p;
    }
}

所以我的 Stream 最终是:

persons.stream()
    .filter(u -> u.size > 12)
    .filter(u -> u.weitght > 12)
    .map(counter)
    .sorted((p1, p2) -> p1.age - p2.age)
    .collect(Collectors.toList())
    .stream()
    .limit((int) (counter.i.intValue() * 0.5))
    .sorted((p1, p2) -> p2.length - p1.length)
    .limit((int) (counter.i.intValue() * 0.5 * 0.2)).forEach((p) -> System.out.println(p));

但我的问题是关于我的示例的另一部分。

collect(Collectors.toList()).stream().

如果我删除该行，结果是当我尝试执行限制时计数器为零。我以某种方式通过使用可变对象来欺骗“有效最终”要求。

我可能错了，但我知道流是先构建的，所以如果我们使用可变对象将参数传递给流中的任何步骤，这些将在创建流时采用。

我的问题是，如果我的假设是正确的，为什么需要这样做？流(如果非并行)可以按顺序通过所有步骤(过滤器、映射..)，因此不需要此限制。

Answer 1

简答

My question is, if my assumption is right, why is this needed? The stream (if non parallel) could be pass sequentially through all the steps (filter, map..) so this limitation is not needed.

如您所知，对于并行流，这听起来很明显:需要此限制，否则结果将是不确定的。

关于非并行流，这是不可能的，因为他们目前的设计:每个项目只被访问一次。如果流确实按照您的建议工作，它们会在进入下一步之前对整个集合执行每个步骤，我认为这可能会对性能产生影响。我怀疑这就是语言设计者做出该决定的原因。

---

为什么在没有 collect 的情况下它在技术上不起作用

你已经知道了，但这里是给其他读者的解释。来自 the docs :

Streams are lazy; computation on the source data is only performed when the terminal operation is initiated, and source elements are consumed only as needed.

Stream 的每一个中间操作，例如filter() 或limit() 实际上只是一些一种初始化流选项的 setter 。

当您调用终端操作时，例如forEach()、collect() 或count()，那是计算发生的时候，按照先前构建的管道处理项目。

这就是为什么在单个项目通过流的第一步之前评估 limit() 的参数。这就是为什么您需要使用终端操作结束流，然后使用 limit() 开始新的流，然后您就会知道。

关于为什么不允许它用于并行流的更详细的答案

让您的流管道成为step X > step Y > step Z。

我们希望并行处理我们的项目。因此，如果我们允许步骤 Y 的行为取决于已经经过 X 的项目，那么 Y 是不确定的。这是因为在项目到达步骤 Y 的那一刻，已经通过 X 的项目集在多次执行中不会相同(因为线程)。

关于为什么不允许它用于非并行流的更详细的答案

根据定义，流用于处理流 中的项目。您可以将非并行流想象如下:一个项目通过所有步骤，然后下一个项目通过所有步骤，依此类推。事实上，文档说明了一切:

The elements of a stream are only visited once during the life of a stream. Like an Iterator, a new stream must be generated to revisit the same elements of the source.

如果流不是这样工作的，那么在进入下一步之前只对整个集合执行每个步骤也好不到哪儿去。这实际上会允许非并行流中的可变参数，但它可能会对性能产生影响(因为我们会在集合上迭代多次)。无论如何，他们目前的行为不允许你想要什么。