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对象完全相同.安德鲁很高兴找到了一篇关于它们如何工作的好文章.

注意:确切的源代码可以在这里找到.