java - 在执行其他操作之前查找 Stream 大小

Question

在我的程序中，我重复¹ collect Java 8 streams将对象的集合减少为单个对象。此集合的大小在整个执行过程中可能变化很大:从 3 个对象到数百个对象。

public void findInterestingFoo(Stream<Foo> foos) {
    internalState.update(foos.collect(customCollector()));
}

在优化我的代码和寻找瓶颈的过程中，我做了流parallel在某一点。这在那个时间点奏效了，因为 Collection 品都相当大。后来，在改变程序的其他部分和参数后，集合变小了。我意识到不使流并行会更有效率。这是有道理的:为 4 个对象在多个线程上分配工作的开销根本不值得。不过，对于数百个对象来说，它是值得的。

如果我只能使大流并行，那将非常方便:

public void findInterestingFoo(Stream<Foo> foos) {
    if (isSmall(foos)) {
        internalState.update(foos.collect(customCollector()));
    } else {
        internalState.update(foos.parallel().collect(customCollector()));
    }
}

当然，当从 an array 创建流时，这可以手动完成, a collection , 或 manually .也就是说，我们知道流中有哪些元素，因此可以对其进行跟踪。然而，我有兴趣以通用方式解决这个问题，因此无论将哪种流传递给 findInterestingFoo，它都会得到适当且尽可能高效的处理。

类似于 count()可能有帮助，除了它在我可以之前终止流 collect

我很清楚流的设计没有固定大小，特别是:

• Possibly unbounded. While collections have a finite size, streams need not. Short-circuiting operations such as limit(n) or findFirst() can allow computations on infinite streams to complete in finite time. — java.util.stream package description

不过，我想知道是否有任何方法可以在对其执行任何操作之前确定流中有多少元素。流真的不知道它是从有限集合创建的吗？

__________
_{¹ 数千次。在我的案例中，优化它使总运行时间从大约 1.5 秒加速到 0.5 秒。}

Answer 1

理论上，你可以这样做:

public void findInterestingFoo(Stream<Foo> foos) {
    Spliterator<Foo> sp = foos.spliterator();
    long size = sp.getExactSizeIfKnown();// returns -1 if not known
          // or sp.estimateSize(); // Long.MAX_VALUE means "unknown"
    internalState.update(
        StreamSupport.stream(sp, size > PARALLEL_THRESHOLD)
                     .collect(customCollector()));
}

spliterator()是消耗输入流的终端操作，但您可以通过 Spliterator至 StreamSupport.stream构造具有完全相同属性的流。第二个参数已经说明流是否应该并行。

理论上。

实际上，当前的流实现将根据流是否并行返回不同的Spliterator 实现。这意味着当原始流在调用 spliterator() 之前尚未并行时，将流重新创建为并行流可能会导致流无法进行并行处理。

但是，如果没有中间操作，例如当您直接传入从集合或数组创建的 Stream 时。

调用 parallel()在 spliterator() 之前获得一个支持并行的流，如果您决定这样做，它仍然可以按顺序运行，在很多情况下都有效。但是，如果输入流中有像 sorted() 这样的有状态中间操作，它们可能会固定为并行运行，即使您按顺序执行 collect(或反之亦然)。

---

另一个问题是基本性质的。元素的数量实际上并没有说明并行处理是否会有好处。这确实取决于每个元素的工作负载，这不仅取决于您的终端 collect 操作，还取决于在进入您的方法之前已经链接到流的操作。即使您断定收集器的工作负载已经足够高，值得并行处理，也可能是传入流具有类似 skip 的操作。 , limit或 distinct (在有序流上)，它通常并行运行更糟并且需要完全不同的阈值。

一个更简单的解决方案是让调用者决定，因为调用者知道流的大小和性质。您甚至不需要向方法的签名添加选项，因为调用者已经可以通过在传递之前调用流上的 parallel() 或 sequential() 来做出决定它到你的方法，你可以通过简单地不改变模式来尊重它。