java - Java Stream API 是如何选择执行计划的？

Question

我刚开始学习 Java 8 中的 Stream API 和一般的函数式编程，但对 Java 并不陌生。我有兴趣了解和了解 Stream API 如何选择执行计划。

它如何知道哪些部分需要并行化，哪些部分不需要？存在多少种执行计划？

基本上，我想知道为什么 Java 8 中的 Streams 有助于使事情变得更快，以及它如何发挥这种“魔力”。

我找不到太多关于这一切如何运作的文献。

Answer 1

这个问题有点宽泛，不好详细解释，但我会尽力回答到满意的程度。我还使用了 ArrayList 的 Stream 示例。

当我们创建流时，返回的对象称为 ReferencePipeline .这个对象可以说是“默认流”对象，因为它还没有任何功能。现在我们必须在惰性方法和急切方法之间做出选择。因此，让我们分别看一个示例。

示例一:filter(Predicate<?>)方法:

filter()方法声明如下:

@Override
public final Stream<P_OUT> filter(Predicate<? super P_OUT> predicate) {
    Objects.requireNonNull(predicate);
    return new StatelessOp<P_OUT, P_OUT>(this, StreamShape.REFERENCE,
                                 StreamOpFlag.NOT_SIZED) {
        @Override
        Sink<P_OUT> opWrapSink(int flags, Sink<P_OUT> sink) {
            return new Sink.ChainedReference<P_OUT, P_OUT>(sink) {
                @Override
                public void begin(long size) {
                    downstream.begin(-1);
                }

                @Override
                public void accept(P_OUT u) {
                    if (predicate.test(u))
                        downstream.accept(u);
                }
            };
        }
    };
}

如您所见，它返回一个 StatelessOp 对象，它基本上是一个新的 ReferencePipeline，其中过滤器评估现在“启用”。换句话说:每次我们向流中添加新的“功能”时，它都会基于旧的流水线创建一个新的流水线，并使用适当的操作标志/方法覆盖。
正如您可能已经知道的那样，在调用急切操作之前不会评估流。所以我们需要一个急切的方法来评估流。

示例二:forEach(Consumer<?>)方法:

@Override
public void forEach(Consumer<? super P_OUT> action) {
    evaluate(ForEachOps.makeRef(action, false));
}

一开始这很短，evaluate()方法只是调用 invoke()方法。在这里了解什么很重要 ForEachOps.makeRef()做。它设置了必要的最后标志并创建了一个 ForEachTask<>这与 ForkJoinTask 完全一样目的。又开心Andrew找到一个不错的paper关于它们的工作原理。

---

注意:可以找到确切的源代码here .