学习Java8的Stream-创新互联

  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  Sink wrapSink(Sink 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) sink;
    }

  copyInto是依次调用Sink.begin，Spliterator.forEachRemaing，Sink.end方法。上文曾经提到，forEachRemaing会以流里的每项数据为参数，层层调用每级Sink的accept。

@Override
    final  void copyInto(Sink wrappedSink, Spliterator spliterator) {
        Objects.requireNonNull(wrappedSink);

        if (!StreamOpFlag.SHORT_CIRCUIT.isKnown(getStreamAndOpFlags())) {
            wrappedSink.begin(spliterator.getExactSizeIfKnown());
            spliterator.forEachRemaining(wrappedSink);
            wrappedSink.end();
        }
        else {
            copyIntoWithCancel(wrappedSink, spliterator);
        }
    }

  至此，整个流的操作，自上而下完全联系到了一起。
  

创新互联www.cdcxhl.cn，专业提供香港、美国云服务器，动态BGP最优骨干路由自动选择，持续稳定高效的网络助力业务部署。公司持有工信部办法的idc、isp许可证， 机房独有T级流量清洗系统配攻击溯源，准确进行流量调度，确保服务器高可用性。佳节活动现已开启，新人活动云服务器买多久送多久。