标签：Pipeline 调用 Stream 学习 Sink 操作 Java8 sink
Stream把对一个集合的很多操作，以一种流式作业串起来，按照类似函数式编程书写代码，界面清爽清晰。
Stream不同于Guava的FluentIterable系列。FluentIterable系列，是通过封装集合（List、Set等）并且重载迭代器、get的方式，进行的transform、filter等，优点是简单并且性能高。缺点是功能单一、并且容易误用。比如，对transform之后的列表的每个项，本质上都是一个视图（View），而不是实际的项（值对象）。多次调用get方法。这本质上每次都会层层调用Function的apply方法，如果其中有复杂运算或者I/O读取，效率是非常低的。
Stream本身，建立了一系列的数据结构，其中的数据都可以看成是实际存在的数据，而并不仅仅是视图。网上分析源代码的文章非常多，这里不重复描述，而是仅仅介绍理解的一个思路。
首先，是几个关键的数据结构。
Stream。接口是BaseStream，可以看做是流基本『形态』的描述。或者说，每个流式作业。虽然后面提到的Pipeline也是实现了BaseStream，但是这样似乎更容易理解。
Pipeline。公共父类是AbstractPipeline。可以看做是流式作业中的每个作业，或者说是每个作业节点。为了防止不必要的装箱和拆箱操作，又分成了ReferencePipeline、IntPipeline、LongPipeline、DoublePipeline。这些管道可以分成三种：头、终结操作、非终结操作。分别对应的类是XXXXPipeline.Head、XXXXPipeline.StatelessOp和XXXXPipeline.StatefulOp、TerminalOp.调用流的工作方法，都会制造出来这样一个Pipeline。比如，调用IntStream.of()，就会生成一个头；调用Stream.map()，会生成一个非终结操作；调用Stream.reduce()，生成一个终结操作。非终结操作里面，都要实现opWrapSink方法，该方法要返回一个Sink。
Sink。每个操作，对应一个Sink。每个Sink，关注三个方法：begin、end、accept。如果当前的Sink没有操作，那么直接调用downstream.accept；否则，把当前操作结果作为参数调用downstream.accept。downstream是什么概念呢？比如，一个流的操作是IntStream.of(1,2,3).map(x -> x + 1).boxed().max()。那么map对应的sink，就是头的downstream，boxed对应的sink就是map的downstream。综合起来，只要调用了头的accept，就会层层调用到最后一个终结操作。终结操作没有opWrapSink方法，所以自然不会调用到后续的流。
Spliterator。流里面数据的访问工具。如果是串行流，一般是直接调用里面的forEachRemaining。该方法里关注action.accept，如果之前串联好了每个Sink，那么这里一句调用，就开始了Sink的层层调用。
至此，基本数据结构就介绍这些。下面，关注流的每个节点，是如何连起来的。
每个Stream，由一组Pipeline节点组成，每追加一个操作，都会向这组Pipeline后面追加一个Pipeline结构。追加时候维护的信息里面，关注sourceStage（头），previousStage（上一个Pipeline），sourceSupplier（流数据的Spliterator），depth（Pipeline长度）。一直到最后的终结操作，连城一个Pipeline链。
对于终结操作，不论是reduce，还是collect操作，都会调用到AbstracePipeline.evaluate方法。以串行流的reduce为例，直接调用到AbstractPipeline.wrapAndCopyInto。
其中，wrap是遍历Pipeline链，调用每个阶段的opWrapSink。这样每个Sink通过方法逐层调用（而非内存数据上的指针链接），从第一个Pipeline的Sink链接到末尾。
@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;
}copyInto是依次调用Sink.begin，Spliterator.forEachRemaing，Sink.end方法。上文曾经提到，forEachRemaing会以流里的每项数据为参数，层层调用每级Sink的accept。
@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);
}
} 至此，整个流的操作，自上而下完全联系到了一起。
标签：Pipeline,调用,Stream,学习,Sink,操作,Java8,sink 来源： https://blog.51cto.com/luckybins/2432442
本站声明： 1. iCode9 技术分享网（下文简称本站）提供的所有内容，仅供技术学习、探讨和分享； 2. 关于本站的所有留言、评论、转载及引用，纯属内容发起人的个人观点，与本站观点和立场无关； 3. 关于本站的所有言论和文字，纯属内容发起人的个人观点，与本站观点和立场无关； 4. 本站文章均是网友提供，不完全保证技术分享内容的完整性、准确性、时效性、风险性和版权归属；如您发现该文章侵犯了您的权益，可联系我们第一时间进行删除； 5. 本站为非盈利性的个人网站，所有内容不会用来进行牟利，也不会利用任何形式的广告来间接获益，纯粹是为了广大技术爱好者提供技术内容和技术思想的分享性交流网站。