JAVA 8 Streams 1

红太狼 2021-05-12 11:55 408阅读 0赞

Java 8中的Stream其实是函数式编程里Monad的概念，关于Monad，可以参考[这篇文章][Link 1]。Monad就是一种设计模式，表示将一个运算过程，通过函数拆解成互相连接的多个步骤，有点链式操作的感觉。先看一个例子：

import java.util.Arrays;
    import java.util.List;
    
    public class Snippet{
        public static void main(String[] args){
            List<String> myList = Arrays.asList("a1", "a2", "b1", "c2", "c1");
    
            myList
                .stream()
                .filter(s -> s.startsWith("c"))       //过滤以c字母开头
                .map(String::toUpperCase)        //字符变成大写
                .sorted()                                     //排序
                .forEach(System.out::println);    //打印输出
        }
    }

Stream 不是集合元素，它不是数据结构并不保存数据，它是有关算法和计算的，它更像一个高级版本的 Iterator。原始版本的 Iterator，用户只能显式地一个一个遍历元素并对其执行某些操作；高级版本的 Stream，用户只要给出需要对其包含的元素执行什么操作，比如 “过滤掉长度大于 10 的字符串”等，Stream 会隐式地在**内部**进行遍历，做出相应的数据转换。

Stream 就如同一个迭代器（Iterator），单向，不可往复，数据只能遍历一次，遍历过一次后即用尽了，就好比流水从面前流过，一去不复返。

而和迭代器又不同的是，Stream 可以并行化操作，迭代器只能命令式地、串行化操作。顾名思义，当使用串行方式去遍历时，每个 item 读完后再读下一个 item。而使用并行去遍历时，数据会被分成多个段，其中每一个都在不同的线程中处理，然后将结果一起输出。Stream 的并行操作依赖于 Java7 中引入的 Fork/Join 框架（JSR166y）来拆分任务和加速处理过程。Java 的并行 API 演变历程基本如下：

*  1.0-1.4 中的 java.lang.Thread
 *  5.0 中的 java.util.concurrent
 *  6.0 中的 Phasers 等
 *  7.0 中的 Fork/Join 框架
 *  8.0 中的 Lambda

当我们使用一个流的时候，通常包括三个基本步骤：

获取一个数据源（source）→ 数据转换→执行操作获取想要的结果，每次转换原有 Stream 对象不改变，返回一个新的 Stream 对象（可以有多次转换），这就允许对其操作可以像链条一样排列，变成一个管道，如下图所示。

![70][]

1）使用Stream的好处

*  对JAVA集合（Collection）对象功能的增强，方便对集合进行各类操作（过滤、求最大值、最小值、统计等）；
 *  更加高效，提供串行和并行两种模式，并行模式利用了Java中的fork/join框架技术，能充分利用多核处理器，提高程序并发性；

2）Stream的特征

*  不是一个数据结构，为lambda表达式设计（所有 Stream 的操作必须以 lambda 表达式为参数）。它没有内部存储，它只是用操作管道从 source（数据结构、数组、generator function、IO channel）抓取数据。它也绝不修改自己所封装的底层数据结构的数据。例如 Stream 的 filter 操作会产生一个不包含被过滤元素的新 Stream，而不是从 source 删除那些元素。
 *  不支持索引访问
 *  很方便的作为数组或集合输出
 *  支持惰性访问（Intermediate都是惰性的）
 *  并行计算
 *  数据源本身可以是无限的（集合有固定大小，Stream 则不必。limit(n) 和 findFirst() 这类的 short-circuiting 操作可以对无限的 Stream 进行运算并很快完成）

3）Stream的操作类型：  
Stream有两种类型的操作：Intermediate操作和Terminal操作。类似于spark中RDD的transform和action两种操作。

*  Intermediate（中间操作）：Stream可以进行多次的Intermediate操作，如前面开头的那个例子，其中filter、map、sorted都是Intermediate操作，注意该操作是惰性化的，当调用到该方法的时候，并没有真正开始Stream的遍历。
 *  Terminal（结束操作）：一个Stream只有一个Terminal操作，如前面开头的那个例子，其中forEach就是Terminal操作，Terminal操作是Stream的最后一个操作，这时候才会开始Stream的遍历。

4）惰性化：

在对于一个 Stream 进行多次转换操作 (Intermediate 操作)，每次都对 Stream 的每个元素进行转换，而且是执行多次，这样时间复杂度就是 N（转换次数）个 for 循环里把所有操作都做掉的总和吗？其实不是这样的，转换操作都是 lazy 的，多个转换操作只会在 Terminal 操作的时候融合起来，一次循环完成。我们可以这样简单的理解，Stream 里有个操作函数的集合，每次转换操作就是把转换函数放入这个集合中，在 Terminal 操作的时候循环 Stream 对应的集合，然后对每个元素执行所有的函数。

5）短路（short-circuiting）：

有些操作不需要处理整个流就能得到结果。例如，假设你需要对一个用 and 连起来的大布尔表达式求值。不管表达式有多长，你只需找到一个表达式为 false ，就可以推断整个表达式将返回 false ，所以用不着计算整个表达式。这就是短路。对于流而言，某些操作（例如 allMatch 、 anyMatch 、 noneMatch 、 findFirst 和 findAny ）不用处理整个流就能得到结果。只要找到一个元素，就可以有结果了。同样， limit 也是一个短路操作：它只需要创建一个给定大小的流，而用不着处理流中所有的元素。在碰到无限大小的流的时候，这种操作就有用了：它们可以把无限流变成有限流。

*  对于一个 intermediate 操作，如果它接受的是一个无限大（infinite/unbounded）的 Stream，但返回一个有限的新 Stream。
 *  对于一个 terminal 操作，如果它接受的是一个无限大的 Stream，但能在有限的时间计算出结果。

当操作一个无限大的 Stream，而又希望在有限时间内完成操作，则在管道内拥有一个 short-circuiting 操作是必要非充分条件。

**1、java.util.stream包结构**：

![70 1][]

1）BaseStream接口：

所有Stream接口类型的父接口，它继承自AutoClosable接口，定义了一些所有Stream都具备的行为。因为继承自AutoClosable接口，所以所有的Stream类型都可以用在Java 7中引入的try-with-resource机制中，以达到自动关闭资源的目的。实际上，只有当Stream是通过Socket，Files IO等方式创建的时候，才需要关闭它。对于来自于Collections，Arrays的Stream，是不需要关闭的。

2）Stream接口：

定义了众多Stream应该具有的行为。最典型的比如filter方法族，map方法族以及reduce方法族，这三个方法是FunctionalProgramming的标志。典型的Map-Filter-Reduce模式便是依靠这三个操作来定义的。

与此同时，Stream接口还定义了一些用于创建Stream的static方法，创建的Stream可以是有限的，也可以是无限的。有限的很好理解，而无限Stream是一个新概念，通过generate方法或者iterate方法实现。

3）IntStream, LongStream 以及 DoubleStream 接口：

基于原生类型int, long以及double的Stream。提供了众多类型相关的操作。典型的例如，sum方法，min/max方法，average方法等。这些方法都是Reduce操作的具体实现。

4）Collect接口：

对于Reduce操作的抽象。此接口中定义了常用的Reduce操作。其中定义的Reduce操作可以通过串行或者并行的方式进行实现。BaseStream接口中的parallel，sequential，unordered方法提供的高层API使并发程序设计变得非常简洁。毕竟，Map-Filter-Reduce模式的灵魂就在于并行计算。

5）Collectors类：

提供了众多可以直接使用的Reduce操作。典型的比如groupingBy以及partitioningBy操作。它们都可以通过串行或者并行的方式进行实现。比如，groupingByConcurrent会使用并行的方式进行grouping操作。

6）StreamSupport类：

提供了底层的一些用于操作Stream的方法，如果不需要创建自己的Stream，一般不需要使用它。

**2、创建Stream对象**：

1）从 Collection 和数组：

*  Collection.stream()
 *  Collection.parallelStream()
 *  Arrays.stream(T array) or Stream.of()

2）从 BufferedReader

*  java.io.BufferedReader.lines()

3）静态工厂

*  java.util.stream.IntStream.range()
 *  java.nio.file.Files.walk()

4）自己创建

*  java.util.Spliterator

5）其它

*  Random.ints()
 *  BitSet.stream()
 *  Pattern.splitAsStream(java.lang.CharSequence)
 *  JarFile.stream()

。。。

参考：

https://www.ibm.com/developerworks/cn/java/j-lo-java8streamapi/

[Link 1]: http://www.ruanyifeng.com/blog/2015/07/monad.html
[70]: /images/20210511/b63bb86f00a845cda8ecede18ffd161a.png
[70 1]: /images/20210511/595338780681425e8b327ac1e2afbc77.png