[三]java8 函数式编程Stream 概念深刻理解 Stream 运行原理 Stream设计思路

Stream的概念定义

 

官方文档是永远的圣经~

 

 

表格内容来自https://docs.oracle.com/javase/8/docs/api/   Package java.util.stream  一节部分原文内容的翻译

int sum = widgets.stream() .filter(b -> b.getColor() == RED) .mapToInt(b -> b.getWeight()) .sum();     流操做被划分为中间和终端操做，并组合成流管道。 一条Stream管道由一个源（如一个集合、一个数组、一个生成器函数或一个i/o通道）组成; 而后是零个或更多的中间操做，例如Stream.filter或者 Stream.map 还有一个终端操做，Stream.forEach or Stream.reduce   中间操做返回一条新流,他们老是惰性的; 执行诸如filter（）之类的中间操做实际上并不会当即执行任何过滤操做，而是建立了一个新流，当遍历时，它包含与给定谓词相匹配的初始流的元素。直到管道的终端操做被执行，管道源的遍历才会开始   终端操做，例如Stream.forEach 和 IntStream.sum，能够遍历流以产生结果或反作用。 在执行终端操做以后，流管道被认为是被消耗掉的，而且不能再被使用; 若是您须要再次遍历相同的数据源，您必须返回到数据源以得到一条新的stream。 在几乎全部状况下，终端操做都很迫切，在返回以前完成了数据源的遍历和管道的处理。只有终端操做iterator() 和 spliterator() 不是;这些都是做为一个“逃生舱口”提供的，以便在现有操做不足以完成任务的状况下启用任意客户控制的管道遍历(我的理解就是若是流不足以提供处理可让你自行遍历处理)   延迟处理流能够显著提升效率; 在像上面的filer-map-sum例子这样的管道中，过滤、映射和求和能够被融合到数据的单个传递中，而且具备最小的中间状态。 惰性还容许在没有必要的状况下避免检查全部数据;对于诸如“查找第一个超过1000个字符的字符串”这样的操做，只须要检查足够的字符串，就能够找到具备所需特征的字符串，而不须要检查源的全部字符串。（当输入流是无限的而不只仅是大的时候，这种行为就变得更加剧要了。）   中间操做被进一步划分为无状态和有状态操做。 无状态操做，如filter和map，在处理新元素时不保留之前处理的元素的状态——每一个元素均可以独立于其余元素的操做处理。有状态的操做，例如distinct和sorted，能够在处理新元素时从先前看处处理的元素中合并状态。   有状态操做可能须要在产生结果以前处理整个输入。 例如，一我的不能从排序流中产生任何结果，直到一我的看到了流的全部元素。 所以，在并行计算下，一些包含有状态中间操做的管道可能须要对数据进行屡次传递，或者可能须要缓冲重要数据。包含彻底无状态的中间操做的管道能够在单次传递过程当中进行处理，不管是顺序的仍是并行的，只有最少的数据缓冲   此外，一些操做被认为是短路操做。一个中间操做，若是在提供无限流输入时，它可能会产生一个有限的流,那么他就是短路的。若是在无限流做为输入时，它可能在有限的时间内终止,这个终端操做是短路的。 在管道中进行短路操做是处理无限流在有限时间内正常终止的必要条件，但不是充分条件

这些流的方法是如何实现的? 类StreamSupport提供了许多用于建立流的低级方法，全部这些方法都使用某种形式的Spliterator. 一个Spliterator.是迭代器Iterator的并行版本 它描述了一个（多是无限的）元素集合，支持顺序前进、批量遍历，并将一部分输入分割成另外一个可并行处理的Spliterator 在最低层，全部的流都由一个spliterator 构造(因此说流就是迭代器的一种高级形式) 在实现Spliterator时，有许多实现选择，几乎全部的实现都是在简单的实现和使用Spliterator流的运行时性能之间进行权衡。建立Spliterator的最简单、但最不高性能的方法是，使用Spliterators.spliteratorUnknownSize(java.util.Iterator, int).虽然这样的Spliterator能够工做，但它极可能提供糟糕的并行性能，由于咱们已经丢失了尺寸信息（底层数据集有多大），而且被限制为一个简单的分割算法。 一个高质量的Spliterator将提供平衡的和知道大小的分割，精确的尺寸信息，以及一些可用于实现优化执行的spliterator 或数据的 characteristics (见Spliterator int characteristics()  ) 可变数据源的Spliterators有一个额外的挑战; 绑定到数据的时间，由于数据可能在建立spliterator的时间和执行流管道的时间之间发生变化。理想状况下，一个流的spliterator 应该报告一个characteristic of IMMUTABLE or CONCURRENT; 若是不是，应该是后期绑定。若是一个源不能直接提供一个推荐的spliterator，它可能会间接地经过Supplier提供一个spliterator,经过接收Supplier做为参数的stream方法构建一个流 public static <T> Stream<T> stream(Supplier<? extends Spliterator<T>> supplier, int characteristics, boolean parallel)   只有在流管道的终端操做开始后，才从supplier处获

 

 

Stream  文档概要译文:

支持顺序和并行聚合操做的一组元素序列 除了Stream 还有专门为原始类型特殊化的IntStream、LongStream和double Stream  全部这些都被称为“流” 集合和流，虽然表面上有一些类似性，但有不一样的设计目的 集合主要关注的是对其元素的有效管理和访问 相比之下，流并无提供直接访问或操纵其元素的方法，而是关注于声明性地描述它们的源和计算操做，这些操做将在该源上进行聚合。 可是，若是所提供的流操做没有提供所需的功能，那么 BaseStream.iterator() 和 BaseStream.spliterator() 操做能够用来执行受控的遍历   示例: widgets 是 Collection<Widget> int sum = widgets.stream() .filter(w -> w.getColor() == RED) .mapToInt(w -> w.getWeight()) .sum();   像上面的“widgets”示例同样，流管道能够看做是在流的数据源上进行的查询。   除非源代码是为并发修改而显式设计的（例如ConcurrentHashMap），不然在查询时 修改流的源 可能致使不可预测或错误的行为。   大多数流操做都接受描述用户指定行为的参数，好比在上面的例子中传递给mapToInt的lambda表达式w-w.getweight（）。 为了保持正确的行为，这些行为参数：         必须是非干扰（也就是它们不修改流源）;         在大多数状况下，必须是无状态的（它们的结果不该该依赖于任何在流水线执行过程当中可能发生变化的状态） 这些参数一般是函数接口的实例，例如Function，通常是lambda表达式或方法引用。除非另有说明，这些参数必须是非空的。   一个流应该只运行一次（调用中间操做或结束操做）。这就排除了好比“forked”流，在这些流中，相同的源提供两个或更多的管道，或者同一流的多个遍历。 一个流实现可能会抛出IllegalStateException 异常，若是它检测到流正在被重用。 然而，因为某些流操做可能返回它们的接收者而不是一个新的stream对象，因此并不能在全部状况下都检测到重用。   Streams有一个BaseStream.close（）方法并实现AutoCloseable，可是几乎全部的stream实例在使用后实际上并不须要关闭。 一般，只有源是IO通道的流（比Files.lines(Path,Charset)）将须要关闭。 大多数流都是由集合、数组或生成函数支持的，这些功能不须要特殊的资源管理。（若是流确实须要关闭，它能够在try-with-resources语句中声明为资源。）   流管道能够按顺序或并行执行 ,这种执行模式是流的属性。 流的类型是建立初始时选择经过顺序或并行操做执行来决定的。（例如，Collections.stream（）建立了一个顺序流，而Collection.parallelStream（）建立了一个并行的流。） 这种执行模式的选择能够由BaseStream.sequential() 或BaseStream.parallel()方法修改，而且可使用BaseStream.isParallel() 方法查询。

 

 

流的基本特色

 

集合是对一组特定类型的元素值序列提供的接口  是数据结构,提供了元素的存取 流也是对一组特定类型元素值序列提供的接口,在于计算,提供了对元素序列的操做计算方式 好比 filter map等

流只能运行一次

流由源 0个或者多个中间操做以及结束操做组成

流操做的方法基本上是函数式接口的实例

流的中间操做是惰性的并不会当即执行 这更有利于内部迭代的优化

流借助于它内部迭代特性提供了声明式的编程方式 更急简洁

中间操做自己会返回一个流,能够将多个操做复合叠加,造成一个更大的流水线

流分为顺序和并行两种方式

 

流与集合主要区别

 

不存储数据

        流不是存储元素的数据结构;相反，它经过一个计算操做的管道，从一个数据源，如数据结构、数组、生成器函数或i/o通道中传递元素

函数特性

        一个流上的操做产生一个结果，可是不会修改它的源。例如，过滤集合 得到的流会产生一个没有被过滤元素的新流，而不是从源集合中删除元素

延迟搜索

        许多流操做，如过滤、映射或重复删除，均可以延迟实现，从而提供出优化的机会。

        例如，“找到带有三个连续元音的第一个字符串”不须要检查全部的输入字符串。

        流操做分为中间（流生成）操做和终端（值或反作用生成）操做。许多的中间操做, 如filter,map等，都是延迟执行。

       中间操做老是lazy的。

Stream多是无界的

        虽然集合的大小是有限的，但流不须要。诸如limit（n）或findFirst（）这样的短路操做能够容许在有限时间内完成无限流的计算。

消耗的

        流的元素只在流的生命周期中访问一次。就像迭代器同样，必须生成一个新的流来从新访问源的相同元素

 

 

能够把流当作一个高级的迭代器Iterator ,内部有它自身运行逻辑的迭代器 你只须要告诉他你想要作什么,他本身就会自动的去迭代筛选组织你想要的数据

 

 

 

Stream的生成

 

目前在java中 集合框架与Stream的结合最多

由于Stream 是对数据项的计算,而集合偏偏是用来存储数据项的数据结构

你固然可使用其余的数据项表示形式

方法列表

• collection.stream   parallelStream  实例方法 转换
• Arrays.stream(Object[]) 静态方法 转换
• Stream类的静态工厂方法 好比  Stream.of(Object[]), IntStream.range(int, int)， Stream.iterate(Object, UnaryOperator)   Stream.generate   
• BufferedReader.lines(); 文件行
• Files类的获取文件路径列表： find(), lines(), list(), walk();
• Random.ints()  随机数流
• JDK中的许多其余流载方法，包括BitSet.stream(), Pattern.splitAsStream(java.lang.CharSequence), and JarFile.stream().

 

经常使用建立方法

 

Stream.of( )      Stream类静态方法      转换XXX为Stream

 

 

Collection stream   集合转换为Stream 特   别   注   意:这是一个default方法,也就意味着若是没有特别处理,全部Collection子类都具备

 

Arrays.stream();     数组转换为Stream

 

 

Stream.iterate   Stream类静态方法 迭代器的形式，建立一个数据流

 

 

好比

Stream.generate

 

好比

 

 

 

Stream类体系结构与流水线设计思路

BaseStream规定了流的基本接口 Stream中定义了map、filter、flatmap等用户关注的经常使用操做； Int~ Long~ Double~是针对于基本类型的特化 方法与Stream中大体对应,固然也有一些差异 BaseStream Stream IntStream LongStream DoubleStream 组建了Java的流体系根基 他们都只是接口

PipelineHelper主要用于Stream执行过程当中相关结构的构建 ReferencePipeline  和 AbstractPipeline 完成了Stream的主要实现 AbstractPipeline类实现了全部的 Stream的中间操做和最终操做 [Int | Long | Double] pipeline 同理相似ReferencePipeline只不过是针对基本类型

 

 

 

 

Head、StatelessOp、StatefulOp为ReferencePipeline中的内部类 [Int | Long | Double]Pipeline 内部也都是定义了这三个内部类

 

说到这咱们已经能够清晰地知道Stream的实现类

 

 

流的族谱

 

 

 

回头看一下获取Stream的方式

好比Collection中调用StreamSupport.stream 返回的是ReferencePipline.Head

 

再好比 Stream的of方法 也仍是调用了StreamSupport.stream方法 他返回的仍是ReferencePipline.Head

 

你会发现流的生成转换建立都是使用StreamSupport StreamSupport 是用于建立和操纵Stream的低级工具方法 除了构造方法,每一个方法都是返回他们对应的Head ReferencePipeline.Head / DoublePipeline.Head / IntPipeline.Head /  LongPipeline.Head

 

 

Stream的操做通常都有三部分构成

数据源

操做( filter map.....)

回调方法(Lambda匿名函数 方法引用)

 

 Stream  使用stage来对操做进行抽象描述

 

 

能够看下Head  StatefulOp  StatelessOp 的源码注释就能够知道了 .分别是 :

Source stage of a ReferencePipeline. 

Base class for a stateful intermediate stage of a Stream.

Base class for a stateless intermediate stage of a Stream.

原始类型特化出来的也是同样

咱们知道Head  StatefulOp  StatelessOp是     ReferencePipeline类型的 ([Int | Long | Double]Pipeline).他们也都是Stream

因此说每一个操做其实也都是Stream

如今也就能够明白为何建立转换生成的流都是Head 了 由于它用来抽象描述 源阶段也就是初始阶段

 

再回头看下AbstractPipeline的概念

“管道”类的抽象基类，它们是流接口及其原始专门化的核心实现。管理管道的建设和评估审查

AbstractPipeline表示一个流管道的初始部分，封装了一个流源和零个或多个中间操做。

单独的AbstractPipeline对象一般用来表示阶段，其中每一个阶段描述的是流源或中间操做。

 

一个具体的中间阶段一般是由AbstractPipeline 构建的--一个特定于操做的具体类,他继承于AbstractPipeline的子类的好比IntPipeline

说白了也就是上面说到过的Head  StatefulOp  StatelessOp  他们自己也是AbstractPipeline类型的

AbstractPipeline包含了评估管道的大部分机制，并实现了操做所使用的方法;

特定类型的类添加了助手方法，用于处理将结果收集到适当的特定类型的容器中。

 

那么AbstractPipeline 是怎么构造管道的呢?

 

先说Head  这是建立生成流的时候返回的对象

能够看得出来 Head的构造也仍是调用的AbstractPipeline

仔细看下他的属性域

private final AbstractPipeline sourceStage;  //反向连接管道链的head,也就是说每一个管道节点都有一个头 private final AbstractPipeline previousStage;//指向上一个 private AbstractPipeline nextStage;//指向下一个 赤裸裸的双向链表

 

 

首先是构造,最终调用AbstractPipeline的构造方法

而后当前节点做为源

上一个为null  下一个为null

 

而后再看下filter和map方法

Stream中将操做抽象化为stage 每一个stage 也就是一个AbstractPipeline 每一个stage 至关于一个双向链表的节点  ,每一个节点都保存Head而后保存着上一个和下一个 这个双向链表就构成了整个流水线 (上面的图看起来next一直是null 是在每一个处理后的this里面的previousStage里面 上一个的next是当前)

 

 

每一个操做向操做同样组合成为双向链表

链表将每一个操做流水线化,可是每一个操做具体的行为是什么?

那么,每一个操做的具体细节又是什么样子的呢?

 

Sink是什么?

sink就是每一个操做具体的行为操做,也能够叫作回调

 

sink是Consumer的扩展,用于流管道中的多个操做阶段中进行数据的监管

经过额外的方法来管理大小信息 流控等

调用accept前须要调用begin通知数据到来,数据发送完成后须要调用end,再次调用accept前必须再次调用begin 

一个sink有两种状态,初始/激活

开始时是初始状态,begin 激活 ,  end使之回到初始状态,能够重复利用 

accept只能在激活状态使用

 

Sink用于协调相邻的stage之间的数据调用

经过begin end accept方法 以及cancellationRequested短路标志位来控制处理流程,对数据进行管控

 

 

再次回头看看filter的代码(理解这个过程须要了解闭包 回调的概念)

 

 

 

每一个操做 statelessOp 或者statefulOp 都是一个AbstractPipeline  也都是stream    也就是流水线的一个抽象stage阶段

AbstractPipeline 他提供了一个opWrapSink方法

statelessOp 或者statefulOp   重写了这个方法

返回了一个Sink对象    (Chained[Reference | Int | Long | Double] 其实都是Sink)

这个对象的accept方法  调用 参数 predicate的test方法  其实也就是调用Lambda 行为化的参数  

这一步至关于封装了当前stage的回调函数

而且他还调用了downstream.accept()方法

 

downstream  是什么?

他就是你传递进去的那个参数 sink

也就是说 statelessOp 或者statefulOp 是流水线 stage的抽象

他有一个opWrapSink()方法,他其中一个参数是Sink

这个方法自己返回一个Sink  sink的accept方法封装了回调函数 也就是当前操做阶段的行为

而后他还会调用参数sink的accept方法

试想,若是传递过来的是下一个操做阶段的sink呢?

思考下上面的这个调用会有什么效果

AbstractPipeline中有一个方法

wrapSink来自于PipelineHelper

从最后一个开始,按照深度进行

如今流水线上从开始到结束的全部的操做都被包装到了一个Sink里  

 

Stream运行流程原理

 

Stream体系是一组接口家族,AbstractPipeline 是接口的实现,PipelineHelper 是管道的辅助类,StreamSupport是流的低级工具类   使用stage来抽象流水线上的每一个操做 其实每一个stage就是一个stream 也就是AbstractPipeline几个子类的  内部子类  Head StatelessOp statefulOp StreamSupport用于建立生成Stream 对应的是Head类 其余的中间操做分为有状态和无状态的,中间操做经过方法好比 filter map 等返回的是StatelessOp  或者 statefulOp  多个stage组合称为双向链表的形式 从而成了整个流水线   有了流水线,相邻两个操做阶段之间如何协调运算? 因而又有了sink的概念,又来协调相邻的stage之间计算运行 他的模式是begin  accept end 还有短路标记 他的accept就是封装了回调方法   因此说每一个操做stage, StatelessOp  或者 statefulOp中又封装了Sink 经过AbstractPipeline提供的opWrapSink方法能够获取这个sink 调用这个sink的accept方法就能够调用当前操做的方法   那么如何串联起来呢?关键点在于opWrapSink方法 ,他接收一个Sink做为参数 在调用accept方法中  能够调用这个入参sink的accept方法 这样子从当前就能调用下一个,也就是说有了推进的动做 那么只须要找到开始,每一个处理了以后都推进下一个,就顺序完成了所欲的操做了   注意上面说的操做都是中间操做,中间操做才会产生操做阶段  终端操做不会增长stage的个数了

 

 

 

Stream预置操做

 

中间操做	filter() flatMap() limit() map()  concat() distinct() peek() skip() sorted() parallel() sequential() unordered()  flatMapTo[Double | Int | Long]  mapTo[ Double | Int | Long ]
结束操做	allMatch() anyMatch() collect() count() findAny() findFirst() forEach() forEachOrdered() max() min() noneMatch() reduce() toArray()

 

操做的参数基本上是函数式接口的实例---->也就是Lambda匿名函数   方法引用

因此说想要使用Stream预置的函数，只须要了解清楚对应的函数式接口便可

 

 

Stream 主要有四类接口：

 

流到流之间的转换 filter(过滤), map(映射转换), mapTo[Int|Long|Double] (到基本类型流的转换), flatMap(流展开合并)，flatMapTo[Int|Long|Double], sorted(排序)，distinct(不重复值)，peek(执行某种操做，流不变，可用于调试)， limit(限制到指定元素数量), skip(跳过若干元素)

流到终值的转换 toArray（转为数组），reduce（推导结果），collect（聚合结果）， min(最小值), max(最大值), count (元素个数)， anyMatch (任一匹配), allMatch(全部都匹配)， noneMatch(一个都不匹配) findFirst（选择首元素），findAny(任选一元素)

直接遍历 forEach (不保证顺序遍历，好比并行流), forEachOrdered（顺序遍历)

构造流 empty (构造空流)，of (单个元素的流及多元素顺序流) iterate (无限长度的有序顺序流)，generate (将数据提供器转换成无限非有序的顺序流)， concat (流的链接)， Builder (用于构造流的Builder对象)

 

 

Stream经常使用API

 

filter 条件筛选   boolean test(T t);

 

 

map   数据转换    R apply(T t); 新的值替换Stream中的值 mapTo[Int | Long | Double] 相似

 

 

 

 

flatMap  R apply(T t); 同map可是是多个流转换为一个流 返回值限定为Stream 其余flatMapTo[Int | Long | Double] 相似

 

 

 

生成流 iterate（） public static<T> Stream<T> iterate(final T seed, final UnaryOperator<T> f)

UnaryOperator  是 Function的一元运算形式

操做单个的操做数，产生的结果同操做数类型，是Function转换数据--针对操做数和结果类型一致的一种特殊类型

好比 x->x+2  就是如此,操做数为x   转换为x+2

Stream.iterate(0, i -> i + 2)

 

生成流 generate() public static<T> Stream<T> generate(Supplier<T> s)

 

 

reduce  归约  原理

一个归约操做（也称为折叠）接受一系列的输入元素，并经过重复应用组合操做将它们组合成一个简单的结果 例如查找一组数字的总和或最大值，或者将元素累积到一个列表中。 流的类中有多种形式的通用归约操做，称为reduce（）和collect（），以及多个专门化的简化形式，如sum（）、max（）或count（）。 固然，这样的操做能够很容易地实现为简单的顺序循环，以下所示： int sum = 0; for (int x : numbers) { sum += x; }     然而，咱们有充分的理由倾向于减小操做，而不是像上面这样的累加运算。 它不只是一个“更抽象的”——它在流上做为一个总体而不是单独的元素来运行——并且一个适当构造的reduce操做本质上是可并行的，只要用于处理元素的函数（s）是结合的和无状态的。举个例子，给定一个数字流，咱们想要找到和，咱们能够写：     int sum = numbers.stream().reduce(0, (x,y) -> x+y); 或者     int sum = numbers.stream().reduce(0, Integer::sum); 这些归约操做几乎不须要修改就能够并行运行     int sum = numbers.parallelStream().reduce(0, Integer::sum);   若是一个操做符或函数 op  知足    (a op b) op c == a op (b op c) ,那么他是结合的 结合性对于并行结算很是重要 好比      a op b op c op d == (a op b) op (c op d)     就能够并行计算  (a op b)    (c op d)   而后再处理他们

 

他是一个结束操做

 

三个重载版本

 

是Function的双参数版本 接受 T  U 转换为R

是BiFunction的特殊化形式,两个输入一个输出,三个参数类型相同

 

Optional<T> reduce(BinaryOperator<T> accumulator) 参数accumulator:  累计计算器——结合两个值的结合性、非干扰、无状态函数 由于没有初始值 因此返回值为 Optional<T> 计算1+2+3+4+5   至关于 boolean foundAny = false; T result = null; for (T element : this stream) { if (!foundAny) { foundAny = true; result = element; } else result = accumulator.apply(result, element); } return foundAny ? Optional.of(result) : Optional.empty();     若是不进入循环,那么foundAny就是false 直接返回 不然第一次循环给result赋值,此后foundAny 为true  result做为一个操做数调用accumulator.apply()   BinaryOperator 意味着两个操做数一个结果数 类型同样 这可不只仅用于累加 还能够用来合并字符串 多种形式

T reduce(T identity, BinaryOperator<T> accumulator); identity 能够理解为初始值, accumulator 同上 你会发现对于上面的例子,可使用Integer::sum 替代其中的lambda 至关于 T result = identity; for (T element : this stream) result = accumulator.apply(result, element) return result;

<U> U reduce(U identity, BiFunction<U, ? super T, U> accumulator, BinaryOperator<U> combiner);   至关于: U result = identity; for (T element : this stream) result = accumulator.apply(result, element) return result;   看得出来和T reduce(T identity, BinaryOperator<T> accumulator);  几乎是同样的 只不过Stream是T 返回是U   可能同样,可能不同了 若是类型相同的话就是彻底同样的了 第三个参数用于在并行计算下 合并各个线程的计算结果 因此要求      combiner.apply(u, accumulator.apply(identity, t)) == accumulator.apply(u, t)     结果不一样  是由于  ((((5+1)+2)+3)+4)+5   和   (5+1)+ (5+2)+ (5+3)+ (5+4)+ (5+5)  运算结果不相同     此处  挨个比较找到最大,和 使用8和每一个数字比较而后在统一比较 的结果是相同的