java-Stream原理及相关操作详解(filter、map、flatMap、peek、reduce、anyMatch等等)

java-Stream原理及相关操作详解

• Stream流前言
• Stream流原理介绍
• Stream-Api常用方法介绍
• • • filter()
    • map()
    • flatMap
    • peek
    • reduce
    • max、min
    • findAny、 findFirst
    • allMatch、anyMatch、noneMatch
    • sorted
    • count

      Stream流前言

      Java8特性主要是Stream流以及函数式接口的出现；本片文章主要对StreamApi中的常用方法进行讲解，再学习Stream流之前个人建议需要对函数式接口要有充分理解并能熟练掌握，对于函数式编程的学习大家可以参考：java 函数式编程（Function、Supplier、Predicate、Consumer）

      （）

      本片文章主要讲解流相关的操作以及部分原理介绍;

      Stream流原理介绍

      Stream流本质上是一个执行计算的过程，将流当中的中间操作组成一个流管道的形式，流管道由源（可能是数组，集合，生成器函数，I / O通道等），零个或多个中间操作 （将流转换为另一个流，如filter(Predicate) ）组成，以及终端操作 （产生结果或副作用，例如count()、forEach(Consumer) 、collect(?); 对于collect()方法来讲是Stream的重中之重，collect()方法的实现依赖于Collector，对于流的终端收集操作(Collector)会放在后续进行讲解；

      （）

      流本身是由三个部分组成，分别是源、中间操作、终止操作；大家先看个例子：

      int sum = widgets.stream()                 .filter(w -> w.getColor() == RED)                 .mapToInt(w -> w.getWeight())                 .sum();

      widgets认为是一个集合本身的引用，通过widgets.stream()则创建了一个源，中间的filter()和mapToInt() 操作是流的中间操作过程，sum是流的一个终端终止操作；

      大家再来观察一下stream()方法；

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

      stream() 是 Collection 中的 default 方法，实际上调用的是StreamSupport.stream() 方法，返回的是 ReferencePipeline.Head的实例；ReferencePipeline表示的是流阶段的中间阶段，ReferencePipeline.Head表示流的源阶段；一个流的只会创建一次源，而流的中间操作可以没有或者多个；ReferencePipeline.Head 的构造函数传递是 ArrayList 中实现的 spliterator 。常用的集合都实现了 Spliterator 接口以支持 Stream。可以这样理解，Spliterator 定义了数据集合流入流水线的方式。

      接下来将对Stream的构造方法也就是流的的创建过程以及中间操作阶段的具体实现进行分析，来进一步理解流的创建以及流是怎么将每一个中间阶段链接起来并且怎么将每一个阶段的输出通知给下游流； 我们先分析stream()方法的最终实现：其实是AbstractPipeline，而ReferencePipeline继承了AbstractPipeline：

          AbstractPipeline(Spliterator source,                     int sourceFlags, boolean parallel) {        //表示上一个阶段，如果这是源阶段，也就是流创建的阶段并没有上游所以这里为null。        this.previousStage = null;        //流管道中的源拆分器；实际行为是和sourceSupplier配合将集合中的元素驶向流        this.sourceSpliterator = source;        //流管道的源阶段        this.sourceStage = this;        // 此管道对象表示的中间操作的操作标志        this.sourceOrOpFlags = sourceFlags & StreamOpFlag.STREAM_MASK;        // The following is an optimization of:        // StreamOpFlag.combineOpFlags(sourceOrOpFlags, StreamOpFlag.INITIAL_OPS_VALUE);        //源和所有操作的组合源标志和操作标志，直到此流水线对象表示的操作为止（包括该流水线对象所代表的操作）。 在管道准备进行评估时有效。        this.combinedFlags = (~(sourceOrOpFlags  previousStage, int opFlags) {        if (previousStage.linkedOrConsumed)            throw new IllegalStateException(MSG_STREAM_LINKED);        // 表示如果该流管道被消费或者链接则为true，上面我们创建了stream()流，调用filter方法则表示我们将filter这个阶段已经连接到流        previousStage.linkedOrConsumed = true;        // 流管道的下一个阶段，上步骤创建好的流对象下一个对象就是当前对象        previousStage.nextStage = this;        // 此时该流管道中的上一个阶段便是源阶段        this.previousStage = previousStage;        this.sourceOrOpFlags = opFlags & StreamOpFlag.OP_MASK;        this.combinedFlags = StreamOpFlag.combineOpFlags(opFlags, previousStage.combinedFlags);        // 源阶段只有一个仍然是previousStage.sourceStage;        this.sourceStage = previousStage.sourceStage;        if (opIsStateful())            sourceStage.sourceAnyStateful = true;        // 新增了一个filter操作，深度便+1；        this.depth = previousStage.depth + 1;    }

      在和源阶段的构造方法和下游流创建的构造方法拿出来对比会发现：

      • previousStage.linkedOrConsumed = true表示上面我们创建了stream()流，调用filter方法则表示我们将filter这个阶段已经连接到流
      • previousStage（上一个阶段） ：对于源的创建来讲没有上一个阶段，该属性为null，对于下游流来讲上一个阶段就是原阶段；
      • linkedOrConsumed （是否消费或链接）：流创建阶段没有被链接或者消费默认为fasle，创建下游流之后被链接上之后为true
      • previousStage.nextStage：上游流的下个阶段，可以理解为源的下一个阶段便是filter
      • depth ，源阶段没有任何中间操作未创建下游流，depth = 0，创建链接之后便+1；

        其实发现previousStage.nextStage = this previousStage.linkedOrConsumed = this; 通过连接上游流和指向下游流发现流的特性是很像一个双向链表

        到目前为止，流的创建（源阶段以及中间阶段）就结束了，接下来我们再看一下ReferencePipeline.Head 代表流的源看看数据是怎么走向的呢？我们来观察一下ReferencePipeline.Head类：

                /**     * Source stage of a ReferencePipeline.     *     * @param  type of elements in the upstream source       * @param  type of elements in produced by this stage     * @since 1.8     */    static class Head extends ReferencePipeline {       //方法目前已经省略便于观看；    }

        其中有两个泛型参数E_IN、E_OUT,显而易见一个表示进，一个表示出，E_IN表示的是上游源的元素的类型，E_OUT表示的是当前阶段所生成的元素类型

        比如我们使用steam: stream().map().filter()… 对于map()阶段来说，E_IN则是stream生成的源(Spliterator)，E_OUT表示map()处理完之后生成的元素，这个元素可能是对象、集合等等。 对于filter()来说 filter()的输入元素就是map()所输出的元素，filter()的输出元素则会作为下一个中间动作的输入元素。。。。

        流的创建以及数据流向已经介绍完了，接下来我们简单介绍流的每一个阶段是怎么进行通知的：Sink

        • Sink 这个接口，它继承自 Consumer 接口，又定义了begin()、end()、cancellationRequested() 方法。Sink 直译过来是水槽，如果把数据流比作水，那水槽就是水会流过的地方。begin() 用于通知水槽的水要过来了，里面会做一些准备工作，同样 end() 是做一些收尾工作。cancellationRequested() 是原来判断是不是可以停下来了。Consumer 里的accept() 是消费数据的地方。

        • 每一个 Sink 都有自己的职责，但具体表现各有不同。无状态操作的 Sink 接收到通知或者数据，处理完了会马上通知自己的 下游。有状态操作的 Sink 则像有一个缓冲区一样，它会等要处理的数据处理完了才开始通知下游，并将自己处理的结果传递给下游。例如 sorted() 就是一个有状态的操作，一般会有一个属于自己的容器，用来记录处自己理过的数据的状态。sorted() 是在执行 begin 的时候初始化这个容器，在执行 accept 的时候把数据放到容器中，最后在执行 end 方法时才正在开始排序。排序之后再将数据，采用同样的方式依次传递给下游节点。最后数据流到终止节点，终止节点将数据收集起来就结束了。

          好了~ ~ 流我们有了一个基础理念、并掌握函数式接口之后我们再来回顾一下Stream中常用方法~~~

          Stream-Api常用方法介绍

          filter()

          Stream filter(Predicate