学习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);
        }
    }

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


文章题目:学习Java8的Stream
本文路径:http://scyanting.com/article/jdpgce.html