Java8流源码解析-串行流

baseStream：对一组可序列化的元素支持串行、并行的聚合操作。
Stream：对一组可序列化的元素支持串行、并行的聚合操作。
PipelineHelper：执行Stream流水线的帮助类，捕获一个Stream流水线的全部信息（输出形状，中间操作，流标记，是否并行等）。
AbstractPipeline：流水线的抽象基本类，提供了流接口的核心实现，管理构造和执行Stream流水线。
ReferencePipeline：对流水线的中间阶段或者源阶段处理的抽象类。
ReferencePipeline.Head：ReferencePipeline源阶段。
ReferencePipeline.StatefulOp：Stream的有状态中间阶段的基本抽象类。
ReferencePipeline.StatelessOp：Stream的无状态中间阶段的基本抽象类。

中间操作和流源都属于ReferencePipeline（即Stream）。

TerminalOp：消费输入的流并产生输出结果。
Sink：Consumer类的扩展，通过流水线的阶段传导值，使用相关方法去管理信息大小、流程控制等。
TerminalSink：一个Sink，用于累积元素被消费的状态，当计算结束后允许去获取计算结果。
ForEachOp：一个终端操作，执行流式线和发送输出结果到自己，作为一个TerminalSink。
ReduceOp：一个终端操作，执行流式线和发送输出结果到AccumulatingSink，RedueceOp用来执行化简操作。
MatchOp：一个短路终端操作，流上的所有元素执行谓词操作，决定是否所有的、任何的、没有元素满足谓词判断。
FindOp：一个短路终端操作，搜索流上的元素，当发现时中断执行。

Sink：Consumer类的扩展，通过流水线的阶段传导值，使用相关方法去管理信息大小、流程控制等。
TerminalSink：一个Sink，用于累积元素被消费的状态，当计算结束后允许去获取计算结果。
FindSink：实现TerminalSink，具有元素发现、当发现目标后请求取消功能。
AccumulatingSink：TerminalSink的一种，实现流上的元素归约和返回计算结果。
Sink.ChainedReference：实现Sink的抽象类，用于创建Sink链。通过begin，end，cancellationRequest连接下游的水槽。
SortedOps.AbstractRefSortingSink：实现Sink的抽象类，实现引用流的排序。
SortedOps.RefSortingSink：实现引用流排序的Sink。
SortedOps.SizedRefSortingSink：实现指定大小的引用流排序的Sink。

拆分器，遍历或者拆分源数据。
拆分器可以用tryAdvance()单个的访问元素或者使用forEachRemaining顺序遍历元素。

Stream流都是获取一个数据源–》数据转换–》执行操作 获取想要的结果。每次转换原有Stream对象不变，返回一个新的stream对象，可以有多次转换，这就允许对流的操作可以向链条一样排列，变成一个流水线/管道（pipeline）。
Stream使用ReferencePipeline记录用户中间操作，它把每一个操作当成一个阶段，它把这个阶段分成三个类型：Head、StatefulOp和StatelessOp，Stream每调用一次操作其实就是生成一个新的阶段。这些阶段通过双向链表的形式组织串联在一起，建立起了阶段的叠加。
阶段叠加起来之后，Stream使用Sink机制把每个操作串联起来。Sink是封装在每个流阶段里面的，包含TerminalSink和ChainedReference，当终止操作执行的时候，内部封装一个accumulatingSink，终止阶段会从TerminalSink开始从下游往上游回溯，其实就是指针迁移，一层一层包装Sink，最终包装出一个Sink链，在Sink链路中，元素是通过accept方法进行发射传递的。构造完成Sink链路之后就可以对每个元素分别执行begin、accept、end操作完成内部迭代了。

Collection类里的stream()方法就是生成流源的：
Collection.java

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

StreamSupport.java

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

可以看到它最终是通过分割迭代器构建了一个流管道的源ReferencePileline.Head，是流水线的开始阶段。通过父类AbstractPipeline属性可以看出流的每个阶段是一个双向链表的节点，而每个阶段都可以反向链接回源阶段。

abstract class AbstractPipeline>
        extends PipelineHelper implements baseStream {
    private static final String MSG_STREAM_linkED = "stream has already been operated upon or closed";
    private static final String MSG_ConSUMED = "source already consumed or closed";

    
    @SuppressWarnings("rawtypes")
    private final AbstractPipeline sourceStage;

    
    @SuppressWarnings("rawtypes")
    private final AbstractPipeline previousStage;

    
    protected final int sourceOrOpFlags;

    
    @SuppressWarnings("rawtypes")
    private AbstractPipeline nextStage;

    
    private int depth;

    
    private int combinedFlags;

    
    private Spliterator sourceSpliterator;

    
    private Supplier> sourceSupplier;

    
    private boolean linkedOrConsumed;

    
    private boolean sourceAnyStateful;

    private Runnable sourceCloseAction;

    
    private boolean parallel;

...

}

构建流源比重很大的一个参数是sourceSpliterator，这个迭代器Spliterator是jdk8引入的接口，类似于Iterator，但是它是可分割的，利用分而治之的思想，在并行流的时候可以利用多线程ForkJoin并行操作，而且每次处理集合元素时使用tryAdvance()或forEachRemaining()方法。
流源的构建负责封装原始数据，并初始化双向链表的数据结构。

以map操作举例。
ReferencePipeline.java

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

方法构建了一个StatelessOp，它是无状态中间操作阶段的基类，该类创建一个双向链表的节点，并把之前的节点设置成上个操作阶段，刚刚创建的节点设置成下一个阶段。
AbstractPipeline.java

   AbstractPipeline(AbstractPipeline 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;
    }

java8串行流源码解析