Stream的使用
首先我们看一下stream的基本使用方法：
ArrayList<String> list = Lists.newArrayList("America", "ABC", "CNN", "OK", "ASYNC"); List<String> strings = list.stream().filter(e -> e.startsWith("A")).map(e -> e + " nice").collect(Collectors.toList());
最终我们会得到ArrayList中以A开头的字母加上“nice”的字符串List，如果放在jdk7里我们会这样写：
ArrayList<String> strings = Lists.newArrayList(); for (String s : list) { if(s.startsWith("A")){ String newStr = s + "nice"; strings.add(newStr); } }
我试着去看源代码，发现Stream实质上就是这样执行我们的需求的。下面就说说我看到了什么。
Stream相关类的介绍
打开java.util.stream包，可以看到核心接口Stream类，顾名思义就是流水的意思，官方文档原话说的是
A sequence of elements supporting sequential and parallel aggregate operations.
Stream就是一个支持串行和并行的聚集操作的一系列元素。
定义了一些中间操作（Intermediate operations）和结束操作（Terminal operations），
中间操作包括无状态(Stateless)操作比如：filter, map, flatMap等，有状态(Stateful)操作比如：distinct, sorted, limit等；
结束操作(Terminal operations)包括非短路操作（short-circuiting）比如：forEach, reduce, collect等和短路操作如：findFirst, findAny；
中间操作不是真正的操作而是一种操作的描述，只有执行到结束操作才会触发实际计算，在结束操作执行之前只是把中间操作记录了下来。无状态中间操作指元素的操作不受其他元素的影响，比如以某一Predicate去filter元素，元素和元素之前不互相影响。而有状态中间操作指的是元素和元素之间是有关联的，比如sorted，只有读取所有元素之后才能确定排序结果。
短路结束操作指的是不用处理所有元素才能返回结果，比如findFirst，只要找到第一个符合条件的元素即可返回结果。非短路结束操作则必须处理完所有元素才能返回结果。
Stream继承了BaseStream，定义了一些Stream的基本操作。
Pipeline记录操作
以上所说的操作需要被按顺序记录下来，这里就需要管道流水线Pipeline的概念来实现。
管道有一个基类PipelineHelper，他是执行Stream管道的一个helper，将Stream的所有信息收集到一个地方。
上面所说的操作其实都定义在PipelineHelper的一个子类ReferencePipeline中，包括Head（Source stage of a ReferencePipeline）、StatelessOp（Base class for a stateless intermediate stage of a Stream.）、StatefulOp（Base class for a stateful intermediate stage of a Stream.）静态内部类。
ReferencePipeline是描述中间操作管道流和源管道流的一个类，同时也实现了Stream接口
在Stream中使用stage（阶段）来描述一个完整的操作，而Head、StatelessOp、StatefulOp这三个操作都是实例化的PipelineHelper，也就是stage。可以把stage理解为带管道的流（Stream with Pipeline）
在本文一开始的例子中，我们分析一下有几个stage，下图：
每一步Stream的方法调用都产生一个新的stage，在随后的分析中会发现，这些stage会以双向链表的方式链接，而每个stage都记录了每一个阶段的操作，这样我们就可以依赖这种数据结构来保存对数据源的所有操作了。
链接stage
stage的链接靠Sink来实现，我们先看一下Sink的接口，我们这里只看ChainedReference
ChainedReference包括：
- begin：在遍历元素前调用，做好遍历准备
- accept：遍历每个元素的时候调用，包含每个stage的操作和回掉函数
- end：遍历结束后调用
- cancellationRequested：是否能够尽早结束遍历，用于短路操作
每个stage都把操作实现在Sink里，上游stage调用下游stage的accept方法，达到按顺序执行每个操作的目的。
stage的自动执行
直接上代码
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); } }; } }; }
上面代码是Stream的filter方法，fiter是一个无状态操作，返回一个新的stage，还实现了AbstractPipeline.opWrapSink来返回stage实现的sink。这里filter的参数是一个predicate，在predicate.test返回true时调用下游的stage的sink的accept方法，这样整个操作流就连续执行下去了。
stage的双向链接
在说Stream自动执行之前，有必要说一说每个stage是怎么链接起来的。Stream在操作时产生的Operation类是如何用双向链表的结构来前后链接的？
在上面Stream.filter的源代码可以看到，filter返回了一个StatelessOp对象，构造函数接受了当前对象this为第一个参数，然后来看StatelessOp的代码：
abstract static class StatelessOp<E_IN, E_OUT> extends ReferencePipeline<E_IN, E_OUT> { /** * Construct a new Stream by appending a stateless intermediate * operation to an existing stream. * * @param upstream The upstream pipeline stage * @param inputShape The stream shape for the upstream pipeline stage * @param opFlags Operation flags for the new stage */ StatelessOp(AbstractPipeline<?, E_IN, ?> upstream, StreamShape inputShape, int opFlags) { super(upstream, opFlags); assert upstream.getOutputShape() == inputShape; } @Override final boolean opIsStateful() { return false; } }
可以看到StatelessOp实现了ReferencePipeline接口，在构造函数里调用了super(upstream, opFlags)，而这个upstream（上游流）参数就是上面传入的this，下游流StatelessOp的upstream就指向this了，这样就通过下游流的upstream链接上游流。目前每个操作之间还只是单链表。
那有人就会想了，下游流保存了上游流的引用，那上游流是怎么保存下游流的引用呢？这就要看最后的结束操作了，我们来看Stream.collect代码：
public final <R, A> R collect(Collector<? super P_OUT, A, R> collector) { A container; if (isParallel() && (collector.characteristics().contains(Collector.Characteristics.CONCURRENT)) && (!isOrdered() || collector.characteristics().contains(Collector.Characteristics.UNORDERED))) { container = collector.supplier().get(); BiConsumer<A, ? super P_OUT> accumulator = collector.accumulator(); forEach(u -> accumulator.accept(container, u)); } else { container = evaluate(ReduceOps.makeRef(collector)); } return collector.characteristics().contains(Collector.Characteristics.IDENTITY_FINISH) ? (R) container : collector.finisher().apply(container); }
这里我们只看串行操作的分支。filter返回了一个结束操作的计算结果。我们来看evaluate方法：
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())); }
AbstractPipeline.evaluate方法接收了一个结束操作对象，我们只看串行操作：
public <P_IN> R evaluateSequential(PipelineHelper<T> helper, Spliterator<P_IN> spliterator) { return helper.wrapAndCopyInto(makeSink(), spliterator).get(); }
继续看AbstractPipeline.wrapAndCopyInto：
@Override final <P_IN, S extends Sink<E_OUT>> S wrapAndCopyInto(S sink, Spliterator<P_IN> spliterator) { copyInto(wrapSink(Objects.requireNonNull(sink)), spliterator); return sink; } @Override final <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); } }
AbstractPipeline.wrapAndCopyInto接收了结束操作的sink，继续看AbstractPipeline.wrapSink:
@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; }
从结束操作的sink开始，一层一层包装sink，最后第一个中间操作的sink在最外层，在每个操作的opWrapSink方法里返回的sink都维护了一个downstream指向后一个操作，这样，双向链表的结构就完成了。这样，我们在copyInto方法里调用begin、accept、end的时候就会通过downstream一层一层的调用下去，最终在结束操作执行实际计算。
结束
Stream的基本原理就分析到这里，希望大家和我一起讨论学习。希望看不明白的同学可以向我提问，看过源码的同学欢迎指出错误！大家一起学习！
