关于java:Java-Stream-源码分析

前言

Java 8 的 Stream 使得代码更加简洁易懂，本篇文章深入分析 Java Stream 的工作原理，并探讨 Steam 的性能问题。

Java 8 汇合中的 Stream 相当于高级版的 Iterator，它能够通过 Lambda 表达式对汇合进行各种十分便当、高效的聚合操作（Aggregate Operation），或者大批量数据操作 (Bulk Data Operation)。

Stream的聚合操作与数据库SQL的聚合操作sorted、filter、map等相似。咱们在应用层就能够高效地实现相似数据库SQL的聚合操作了，而在数据操作方面，Stream不仅能够通过串行的形式实现数据操作，还能够通过并行的形式解决大批量数据，进步数据的解决效率。

操作分类

官网将 Stream 中的操作分为两大类：

• 两头操作（Intermediate operations），只对操作进行了记录，即只会返回一个流，不会进行计算操作。
• 终结操作（Terminal operations），实现了计算操作。

两头操作又能够分为：

• 无状态（Stateless）操作，元素的解决不受之前元素的影响。
• 有状态（Stateful）操作，指该操作只有拿到所有元素之后能力继续下去。

终结操作又能够分为：

• 短路（Short-circuiting）操作，指遇到某些符合条件的元素就能够失去最终后果
• 非短路（Unshort-circuiting）操作，指必须解决完所有元素能力失去最终后果。

操作分类详情如下图所示：

源码构造

Stream 相干类和接口的继承关系如下图所示：

BaseStream

最顶端的接口类，定义了流的根本接口办法，最次要的办法为 spliterator、isParallel。

Stream

最顶端的接口类。定义了流的罕用办法，例如 map、filter、sorted、limit、skip、collect 等。

ReferencePipeline

ReferencePipeline 是一个构造类，定义外部类组装了各种操作流，定义了Head、StatelessOp、StatefulOp三个外部类，实现了 BaseStream 与 Stream 的接口办法。

Sink

Sink 接口定义了 Stream 之间的操作行为，蕴含 begin()、end()、cancellationRequested()、accpt()四个办法。ReferencePipeline最终会将整个 Stream 流操作组装成一个调用链，而这条调用链上的各个 Stream 操作的高低关系就是通过 Sink 接口协议来定义实现的。

操作叠加

Stream 的根底用法就不再叙述了，这里从一段代码开始，剖析 Stream 的工作原理。

@Testpublic void testStream() {    List<String> names = Arrays.asList("kotlin", "java", "go");    int maxLength = names.stream().filter(name -> name.length() <= 4).map(String::length)            .max(Comparator.naturalOrder()).orElse(-1);    System.out.println(maxLength);}

当应用 Stream 时，次要有 3 局部组成，上面一一解说。

加载数据源

调用 names.stream() 办法，会首次加载 ReferencePipeline 的 Head 对象，此时为加载数据源操作。

java.util.Collection#stream

default Stream<E> stream() {    return StreamSupport.stream(spliterator(), false);}

StreamSupport 类中的 stream 办法，初始化了一个 ReferencePipeline的 Head 外部类对象。

java.util.stream.StreamSupport#stream(java.util.Spliterator<T>, boolean)

public static <T> Stream<T> stream(Spliterator<T> spliterator, boolean parallel) {    Objects.requireNonNull(spliterator);    return new ReferencePipeline.Head<>(spliterator,                                        StreamOpFlag.fromCharacteristics(spliterator),                                        parallel);}

两头操作

接着为 filter(name -> name.length() <= 4).mapToInt(String::length)，是两头操作，分为无状态两头操作 StatelessOp 对象和有状态操作 StatefulOp 对象，此时的 Stage 并没有执行，而是通过AbstractPipeline 生成了一个两头操作 Stage 链表。

java.util.stream.ReferencePipeline#filter

@Overridepublic 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);                }            };        }    };}

java.util.stream.ReferencePipeline#map

@Override@SuppressWarnings("unchecked")public final <R> Stream<R> map(Function<? super P_OUT, ? extends R> mapper) {    Objects.requireNonNull(mapper);    return new StatelessOp<P_OUT, R>(this, StreamShape.REFERENCE,                                    StreamOpFlag.NOT_SORTED | StreamOpFlag.NOT_DISTINCT) {        @Override        Sink<P_OUT> opWrapSink(int flags, Sink<R> sink) {            return new Sink.ChainedReference<P_OUT, R>(sink) {                @Override                public void accept(P_OUT u) {                    downstream.accept(mapper.apply(u));                }            };        }    };}

能够看到 filter 和 map 办法都返回了一个新的 StatelessOp 对象。new StatelessOp 将会调用父类 AbstractPipeline 的构造函数，这个构造函数将前后的 Stage 分割起来，生成一个 Stage 链表：

AbstractPipeline(AbstractPipeline<?, E_IN, ?> previousStage, int opFlags) {    if (previousStage.linkedOrConsumed)        throw new IllegalStateException(MSG_STREAM_LINKED);    previousStage.linkedOrConsumed = true;    previousStage.nextStage = this;    this.previousStage = previousStage;    this.sourceOrOpFlags = opFlags & StreamOpFlag.OP_MASK;    this.combinedFlags = StreamOpFlag.combineOpFlags(opFlags, previousStage.combinedFlags);    this.sourceStage = previousStage.sourceStage;    if (opIsStateful())        sourceStage.sourceAnyStateful = true;    this.depth = previousStage.depth + 1;}

终结操作

最初为 max(Comparator.naturalOrder())，是终结操作，会生成一个最终的 Stage，通过这个 Stage 触发之前的两头操作，从最初一个Stage开始，递归产生一个Sink链。

java.util.stream.ReferencePipeline#max

@Overridepublic final Optional<P_OUT> max(Comparator<? super P_OUT> comparator) {    return reduce(BinaryOperator.maxBy(comparator));}

最终调用到 java.util.stream.AbstractPipeline#wrapSink，这个办法会调用 opWrapSink 生成一个 Sink 链表，对应到本文的例子，就是 filter 和 map 操作。

@Override@SuppressWarnings("unchecked")final <P_IN> Sink<P_IN> wrapSink(Sink<E_OUT> sink) {    Objects.requireNonNull(sink);    for ( @SuppressWarnings("rawtypes") AbstractPipeline p=AbstractPipeline.this; p.depth > 0; p=p.previousStage) {        sink = p.opWrapSink(p.previousStage.combinedFlags, sink);    }    return (Sink<P_IN>) sink;}

在下面 opWrapSink 上断点调试，发现最终会调用到本例中的 filter 和 map 操作。

wrapAndCopyInto 生成 Sink 链表后，会通过 copyInfo 办法执行 Sink 链表的具体操作。

@Overridefinal <P_IN> void copyInto(Sink<P_IN> wrappedSink, Spliterator<P_IN> spliterator) {    Objects.requireNonNull(wrappedSink);    if (!StreamOpFlag.SHORT_CIRCUIT.isKnown(getStreamAndOpFlags())) {        wrappedSink.begin(spliterator.getExactSizeIfKnown());        spliterator.forEachRemaining(wrappedSink);        wrappedSink.end();    }    else {        copyIntoWithCancel(wrappedSink, spliterator);    }}

下面的外围代码是：

spliterator.forEachRemaining(wrappedSink);

java.util.Spliterators.ArraySpliterator#forEachRemaining

@Overridepublic void forEachRemaining(Consumer<? super T> action) {    Object[] a; int i, hi; // hoist accesses and checks from loop    if (action == null)        throw new NullPointerException();    if ((a = array).length >= (hi = fence) &&        (i = index) >= 0 && i < (index = hi)) {        do { action.accept((T)a[i]); } while (++i < hi);    }}

断点调试，能够发现首先进入了 filter 的 Sink，其中 accept 办法的入参是 list 中的第一个元素“kotlin”（代码中的 3 个元素是："kotlin", "java", "go"）。filter 的传入是一个 Lambda 表达式：

filter(name -> name.length() <= 4)

显然这个第一个元素“kotlin”的 predicate 是不会进入的。

对于第二个元素“java”，predicate.test 会返回 true（字符串“java”的长度<=4），则会进入 map 的 accept 办法。

本次调用 accept 办法时，empty 为 false，会将 map 后的后果（int 类型的 4）赋值给 t。

public static <T> TerminalOp<T, Optional<T>>makeRef(BinaryOperator<T> operator) {    Objects.requireNonNull(operator);    class ReducingSink            implements AccumulatingSink<T, Optional<T>, ReducingSink> {        private boolean empty;        private T state;        public void begin(long size) {            empty = true;            state = null;        }        @Override        public void accept(T t) {            if (empty) {                empty = false;                state = t;            } else {                state = operator.apply(state, t);            }        }        ……        }}

对于第三个元素“go”，也会进入 accept 办法，此时 empty 为 true, map 后的后果（int 类型的 2）会与上次的后果 4 通过自定义的比拟器相比拟，存入合乎后果的值。

public static <T> BinaryOperator<T> maxBy(Comparator<? super T> comparator) {    Objects.requireNonNull(comparator);    return (a, b) -> comparator.compare(a, b) >= 0 ? a : b;}

本文代码中的 max 传入的比拟器为：

max(Comparator.naturalOrder())

至此会返回 int 类型的 4。

并行处理

下面的例子是串行解决的，如果要改成并行也很简略，只须要在 stream() 办法后加上 parallel() 就能够了，并行代码能够写成：

@Testpublic void testStream() {    List<String> names = Arrays.asList("kotlin", "java", "go");    int maxLength = names.stream().parallel().filter(name -> name.length() <= 4)            .map(String::length).max(Comparator.naturalOrder()).orElse(-1);    System.out.println(maxLength);}

Stream 的并行处理在执行终结操作之前，跟串行解决的实现是一样的。而在调用终结办法之后，实现的形式就有点不太一样，会调用 TerminalOp 的 evaluateParallel 办法进行并行处理。

final <R> R evaluate(TerminalOp<E_OUT, R> terminalOp) {    assert getOutputShape() == terminalOp.inputShape();    if (linkedOrConsumed)        throw new IllegalStateException(MSG_STREAM_LINKED);    linkedOrConsumed = true;    return isParallel()            ? terminalOp.evaluateParallel(this, sourceSpliterator(terminalOp.getOpFlags()))            : terminalOp.evaluateSequential(this, sourceSpliterator(terminalOp.getOpFlags()));}

外围是应用了 ForkJoin 框架，对 Stream 解决进行分片，最终会调用上面的代码，这里就不开展剖析了。

java.util.stream.AbstractTask#compute

@Overridepublic void compute() {    Spliterator<P_IN> rs = spliterator, ls; // right, left spliterators    long sizeEstimate = rs.estimateSize();    long sizeThreshold = getTargetSize(sizeEstimate);    boolean forkRight = false;    @SuppressWarnings("unchecked") K task = (K) this;    while (sizeEstimate > sizeThreshold && (ls = rs.trySplit()) != null) {        K leftChild, rightChild, taskToFork;        task.leftChild  = leftChild = task.makeChild(ls);        task.rightChild = rightChild = task.makeChild(rs);        task.setPendingCount(1);        if (forkRight) {            forkRight = false;            rs = ls;            task = leftChild;            taskToFork = rightChild;        }        else {            forkRight = true;            task = rightChild;            taskToFork = leftChild;        }        taskToFork.fork();        sizeEstimate = rs.estimateSize();    }    task.setLocalResult(task.doLeaf());    task.tryComplete();}

并行谬误的应用办法

@Testpublic void testParallelWrong() {    List<Integer> parallelList = new ArrayList<>();    IntStream.range(0, 1000).boxed().parallel().filter(i -> i % 2 == 1)            .forEach(parallelList::add);    System.out.println(parallelList.size());}

下面的输入后果会常常小于500，这是因为 parallelList 的类型是 ArrayList，并不是线程平安的，在执行 add 操作时，可能正好赶上扩容或者线程被占用，会笼罩其余线程的赋好的值。

并行正确的应用办法

@Testpublic void testParallelRight() {    List<Integer> parallelList = IntStream.range(0, 1000).boxed().parallel()            .filter(i -> i % 2 == 1).collect(Collectors.toList());    System.out.println(parallelList.size());}

性能

上面的文章参考自：JavaLambdaInternals/8-Stream Performance.md，侵删。

为保障测试后果真实可信，咱们将JVM运行在-server模式下，测试数据在GB量级，测试机器采纳常见的商用服务器，配置如下：