[大数据]Java 基础 —— Java 8 新特性 stream 与 Lambda

一、Stream 简介

Java 8 引入了全新的 Stream API，这里的 Stream 和 I/O 流不同，它更像具有 Iterable 的集合类，但行为和集合类又有所不同。Stream 是 Java 8 的新特性，是对容器对象功能的增强，它专注于对容器对象进行各种非常便利、高效的聚合操作（aggregate operation）或者大批量数据操作。Stream 是用函数式编程方式在集合类上进行复杂操作的工具，开发者可以更容易地使用 Lambda 表达式，并且更方便地实现对集合的查找、遍历、过滤以及常见计算等。同时，它提供串行和并行两种模式进行汇聚操作，并发模式能够充分利用多核处理器的优势，使用 fork/join 并行方式来拆分任务和加速处理过程。所以说，Java 8 中首次出现的 java.util.stream 是一个函数式语言+多核时代综合影响的产物。

Stream 特点：

（1）Stream 流是数据渠道，用于操作数据源（集合、数组等）所生成的元素序列；
（2）Stream 相当于高级版的 Iterator；
（3）Stream 的聚合操作与数据库 SQL 的聚合操作 sorted、filter、map 等非常类似；
（4）Stream 是对容器对象功能的增强，它专注于对容器对象进行各种非常便利、高效的 聚合操作（aggregate operation）或者大批量数据操作；
（5）Stream 是用函数式编程方式在集合类上进行复杂操作的工具，其集成了Java 8中的众多新特性之一的聚合操作，开发者可以更容易地使用Lambda表达式，并且更方便地实现对集合的查找、遍历、过滤以及常见计算等；
（6）在数据操作方面，Stream 不仅可以通过串行的方式实现数据操作，还可以通过并行的方式处理大批量数据，提高处理效率；

Stream 注意点：

（1）Stream 不会自己存储数据。
（2）Stream 不会改变原对象，他们会返回一个新的 Stream。
（3）Stream 操作是延迟的，他们会等到需要的结果时才执行。
（4）用并行流并不一定会提高效率，因为 jvm 对数据进行切片和切换线程也是需要时间的。

1、什么是聚合操作

在传统的 J2EE 应用中，Java 代码就必须依赖于关系型数据库的聚合操作来完成诸如客户每月平均消费金额、最昂贵的在售商品、本周完成的有效订单（排除了无效的）以及取十个数据样本作为首页推荐等需求。但在当今这个数据大爆炸的时代，在数据来源多样化、数据海量化的今天，很多时候不得不脱离 RDBMS，或者以底层返回的数据为基础进行更上层的数据统计。
而 Java 的集合 API 中，仅仅有极少量的辅助型方法，更多的时候是程序员需要用 Iterator 来遍历集合，完成相关的聚合应用逻辑，这是一种远不够高效、笨拙的方法。而在 Java 8 使用 Stream，代码更加简洁易读；而且使用并发模式，程序执行速度更快。

2、Stream 流

（1）什么是流

1）Stream 不是集合元素，它不是数据结构并不保存数据，它是有关算法和计算的，它更像一个高级版本的 Iterator。原始版本的 Iterator，用户只能显式地一个一个遍历元素并对其执行某些操作；高级版本的 Stream，用户只要给出需要对其包含的元素执行什么操作，比如，“过滤掉长度大于 10 的字符串”、“获取每个字符串的首字母”等，Stream 会隐式地在内部进行遍历，做出相应的数据转换。Stream 就如同一个迭代器（Iterator），单向不可往复，数据只能遍历一次，遍历过一次后即用尽了，就好比流水从面前流过，一去不复返。

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

3）Stream 的另外一大特点是，数据源本身可以是无限的。

（2）流的构成

当我们使用一个流的时候，通常包括三个基本步骤：获取数据源（source）→ 数据转换 → 执行操作获取想要的结果。如下图所示：

（3）stream 流的两种操作

每次数据转换都保持原有 Stream 对象不改变，返回一个新的 Stream 对象（可以有多次转换），这就允许对其操作可以像链条一样排列，变成一个管道，如：

图中所示，整个过程就是将 goods 元素集合作为一个 “序列”，进行一组 “流水线” 操作： goods 集合提供了元素序列的数据源；通过 stream() 方法获得 Stream，filter / sorted / limit 是一组链式数据处理，连接起来”构成 “流水线”；forEach 最终执行。

1）filter/sorted/limit 的返回值均为 Stream（类似于 Builder 模式），但它们并不立即执行，而是构成了 “流水线”，直到 forEach 最终执行，并且关闭 Stream。 因此将 filter/sorted/limited 等能够 “连接起来”，并且返回 Stream 的方法称为 “中间操作”（Intermediate）；将 forEach 等最终执行，并且关闭 Stream 的方法称为 “终止操作”（Terminal）。
2）intermediate 操作目的主要是打开流，做出某种程度的数据映射/过滤，然后返回一个新的流，交给下一个操作使用。这类操作都是惰性化的（lazy），就是说，仅仅调用到这类方法，并没有真正开始流的遍历。一个流只能有一个 terminal 操作，当这个操作执行后，流就被使用“光”了，无法再被操作。所以这必定是流的最后一个操作。Terminal 操作的执行，才会真正开始流的遍历，并且会生成一个结果。
3）Stream 的中间操作并不是立即执行，而是 “延迟的”、“按需计算”；完成 “终止操作” 后，Stream 将被关闭。多个中间操作可以连接起来性格一条流水线，除非流水线上触发器终止操作，否则中间操作不会执行任何的处理，而是在终止操作时一次性全部处理，成为惰性求值。

（4）惰性求值和及早求值方法

1）像 filter 这样只描述Stream，最终不产生新集合的方法叫作惰性求值方法；而像 count 这样最终会从 Stream 产生值的方法叫作及早求值方法。

如何判断一个操作是惰性求值还是及早求值：
只需要看其返回值即可：如果返回值是Stream，那么就是惰性求值；如果返回值不是 Stream 或者是 void，那么就是及早求值。

2）在一个 Stream 操作中，可以有多次惰性求值，但有且仅有一次及早求值。
3）中间操作是 lazy(惰性求值) 的，只有在终止操作执行时，才会一次性执行。可以这么认为，Stream 里有个操作函数的集合，每次转换操作就是把转换函数放入这个集合中，在终止操作的时候循环 Stream 对应的集合，然后对每个元素执行所有的函数。

（5）Stream 流的并行

并行处理指的是 Stream 结合了 ForkJoin 框架，对 Stream 处理进行了分片，Spliterator.estimateSize 会估算出分片的数据量；
通过预估的数据量获取最小处理单元的阈值，如果当前分片大小大于最小处理单元的阈值，就继续切分集合；
每个分片都将会生成一个 Sink 链表，当所有分片操作完成后，ForkJoin 框架将会合并所有分片结果集。

二、Lambda 表达式

Lambda 表达式是 Java 8 中提供的一种新的特性，它支持 Java 能进行简单的“函数式编程”。 它是一个匿名函数，即没有函数名的函数。
Lambda 表达式允许把函数作为一个方法的参数（函数作为参数传递进方法中)。

1、函数式接口

如果定义的一个接口有且只有一个抽象方法 ，这样的接口就成为函数式接口（Functional Interface）。函数式接口可以有任意个 default 或着 static 方法。

函数式接口有时候是说具有单个抽象方法的接口， 或着 SAM interfaces。

示例：

interface Foo1 {
    void bar();
}

interface Foo2 {
	static String bar() { 	// static 方法
        return "baz";
    }
    int bar(boolean baz);
}

interface Foo3 {
    default String bar() { 	// default 方法
        return "baz";
    }
    void quux();
}

interface Foo4 {
    static String bar1() { 	// static 方法
        return "baz1";
    }
    default String bar2() { 	// default 方法
        return "baz2";
    }
    void quux();
}

2、Lambda 表达式使用时机

任何函数式接口都可以使用 lambda 表达式替换。 lambda 表达式只能出现在目标类型为函数式接口的上下文中，在可以使用 lambda 表达式的地方，方法声明时必须包含一个函数式接口。
Lambda 表达式，其本质即为实现函数式接口的一种方式。

应用场景：

你在某处就真的只需要一个能做一件事情的函数而已，连它叫什么名字都无关紧要。
Lambda 表达式就可以用来做这件事。

3、Lambda 表达式语法

参数 —> 带返回值的表达式/无返回值的处理

(parameters) -> expression      // 表达式
(parameters) ->{ 				// 语句块
				statements; 
				} 

1）Lambda 表达式使用 () 表示没有参数。
2）如果 Lambda 表达式中只包含一个参数，可省略掉 () 。
3）如果 Lambda 表达式的主题是一段代码块，需要使用 {} ，该代码块与普通的Java代码块并无区别，也可以返回或抛出异常。
4）如果参数的类型可以由编译器推断得出可以省略参数类型，当然你也可以加上。

示例：

//1. 接收2个int型整数,返回他们的和
(int x, int y) -> x + y;

//2. 接受一个 string 对象,并在控制台打印,不返回任何值(看起来像是返回void)
(String s) -> System.out.print(s);

4、Lambda 表达式实现函数式接口

在 Java 8 以前的代码中，为了实现带一个方法的接口，往往需要定义一个匿名类并复写接口方法，代码显得很臃肿。

在 Lambda 出现之前，如果我们需要实现一个函数式接口可能需要下面这种方式：

Foo1 foo1 = new Foo1() {
      @Override
      public void bar() {
        System.out.println("Hello bar");
      }
    };

foo1.bar();

在Java 8中，对于只有一个方法的接口可以把它视为一个函数。实际上，lambda 表达式最终也被编译为一个实现类，不过语法上做了简化。
如果改成使用 Lambda 实现函数式接口就会简单许多：

Foo1 foo1 = ()->System.out.println("Hello Lambda!~");
...
foo1.bar();

5、Lambda 表达式特性

（1）类型推导

编译器负责推导 lambda 表达式的类型。它利用 lambda 表达式所在上下文所期待的类型进行推导， 这个被期待的类型被称为目标类型。就是说我们传入的参数可以无需指定类型！

（2）变量捕获

在 Java 7 中，编译器对内部类中引用的外部变量（即捕获的变量）要求非常严格： 如果捕获的变量没有被声明为 final 就会产生一个编译错误。 我们现在放宽了这个限制——对于 lambda 表达式和内部类， 我们允许在其中捕获那些符合有效只读（Effectively final）的局部变量。
简单的说，如果一个局部变量在初始化后从未被修改过，那么它就符合有效只读的要求， 换句话说，加上 final 后也不会导致编译错误的局部变量就是有效只读变量。
注意：此处和final关键字一样，指的是引用不可改！（感觉没多大意义，还不是用的final）

（3）方法引用

如果我们想要调用的方法拥有一个名字，我们就可以通过它的名字直接调用它。

Comparator byName = Comparator.comparing(Person::getName); 

此处无需再传入参数，lambda 会自动装配成 Person 类型进来然后执行 getName() 方法，而后返回 getName() 的 String。

方法引用有很多种，它们的语法如下：

1）静态方法引用：ClassName::methodName
2）实例上的实例方法引用：instanceReference::methodName
3）超类上的实例方法引用：super::methodName
4）类型上的实例方法引用：ClassName::methodName
5）构造方法引用：Class::new
6）数组构造方法引用：TypeName[]::new

（4）Java 提供给我们的 SAM 接口

Java 8中增加了一个新的包：java.util.function，它里面包含了常用的函数式接口，例如：

1）Predicate——接收T对象并返回boolean
2）Consumer——接收T对象，不返回值
3）Function<T, R>——接收T对象，返回R对象
4）Supplier——提供T对象（例如工厂），不接收值
5）UnaryOperator——接收T对象，返回T对象
6）BinaryOperator——接收两个T对象，返回T对象

6、@FunctionalInterface 注解

@FunctionalInterface 是 Java 8 新加入的一种接口，用于指明该接口类型声明是根据 Java 语言规范定义的函数式接口。我们常用的一些接口 Callable、Runnable、Comparator 等在 JDK8 中都添加了 @FunctionalInterface 注解。

@FunctionalInterface 注解的使用规则如下：

（1）该注解只能标记在"有且仅有一个抽象方法"的函数式接口上。

（2）JDK8 接口中的静态方法（sattic）和默认方法（default），都不算是抽象方法。

（3）接口默认继承 java.lang.Object，所以如果接口显示声明覆盖 Object 中的方法，那么也不算抽象方法。

（4）该注解不是必须的，如果一个接口符合"函数式接口"定义，那么加不加该注解都没有影响，加上该注解能够更好地让编译器进行检查。如果编写的不是函数式接口，但是加上了 @FunctionInterface，那么编译器会报错。

7、实例

（1）自定义 SAM 接口

public interface MyActionInterface {

    public void saySomeThing(String str);
    
     // Java8引入的新特性 接口中可以定义一个default方法的实现 (不是abstract)
    default void say(){
        System.out.println("default say");
    }
}

（2）使用 lambda 表达式

public class WantSay {
    /**
     * 执行接口中的saySomeThing方法
     * @param action
     * @param thing
     */
    public static void excuteSay(MyActionInterface action,String thing){
        action.saySomeThing(thing);
    }

    public static void main(String[] args) {
		// 第一个参数使用 Lambda 实现的接口
        excuteSay((String s) -> System.out.println(s),"Hello world new");
    }
}

（3）使用方法引用

public class Person {
    private String name;      //姓名
    private String location;  //地址

    public String getName() {
        return name;
    }
    public void setName(String name) {
        this.name = name;
    }
    public String getLocation() {
        return location;
    }
    public void setLocation(String location) {
        this.location = location;
    }

    @Override
    public String toString() {
        // TODO Auto-generated method stub
        return "Person:" + name + "," + location;
    }
}

//使用String默认的排序规则，比较的是Person的name字段
Comparator<Person> byName = Comparator.comparing(p -> p.getName());
//不用写传入参数,传入的用Person来声明
Comparator<Person> byName2 = Comparator.comparing(Person::getName);

三、Stream 使用

stream 的方法里面大多都使用了 lambda 表达式。
stream 就是 Java 8 提供给我们的对于元素集合统一、快速、并行操作的一种方式。 它能充分运用多核的优势，以及配合 Lambda 表达式、链式结构对集合等进行许多有用的操作。

1、创建 Stream

（1）从 Collection 和数组获得

Collection 接口和数组的默认方法（默认方法是Java的新特性之一），把一个 Collection 对象转换成 Stream。

// 1、通过 Collection 系列提供的 stream()（串行） 或 parallelStream()（并行）获取
Collection.stream();	// 转化成串行流 
Collection.parallelStream();	// 转化成并行流 
// 2、通过 Arrays 中的静态方法 stream()  获取数据流
Arrays.stream(T array); 	

注意：

对于基本数值型，目前有三种对应的包装类型 Stream：IntStream、LongStream、DoubleStream。当然我们也可以用 Stream<Integer>、Stream<Long> 和 Stream<Double>，但是 boxing/unboxing 会很耗时，所以特别为这三种基本数值型提供了对应的Stream。
Java 8 中还没有提供其它数值型 Stream，因为这将导致扩增的内容较多。而常规的数值型聚合运算可以通过上面三种 Stream 进行。

（2） Stream 接口的静态工厂

注意：Java 8 里接口可以带静态方法。

1）of() 方法

其生成的 Stream 是有限长度的，Stream 的长度为其内的元素个数。

Stream.of(T... values)	// 返回含有多个T元素的 
Stream.of(T t)	// 返回含有一个T元素的 Stream 

示例

Stream<Integer> integerStream = Stream.of(1, 2, 3);
Stream<String> stringStream = Stream.of("A");

2）generator() 方法

generator() 方法返回一个无限长度的Stream，其元素由Supplier接口的提供。在 Supplier 是一个函数接口，只封装了一个 get() 方法，其用来返回任何泛型的值，该结果在不同的时间内，返回的可能相同也可能不相同，没有特殊的要求。

Stream.generate(Supplier<T> s)	// 返回一个无限长度的Stream 

这种情形通常用于随机数、常量的 Stream，或者需要前后元素间维持着某种状态信息的 Stream。把 Supplier 实例传递给 Stream.generate() 生成的 Stream，默认是串行（相对 parallel 而言）但无序的（相对 ordered 而言）。
示例：

Stream<Double> generateA = Stream.generate(new Supplier<Double>() {
    @Override
    public Double get() {
        return java.lang.Math.random();
    }
  });

Stream<Double> generateB = Stream.generate(()-> java.lang.Math.random()); 
Stream<Double> generateC = Stream.generate(java.lang.Math::random); 

3）iterate() 方法

iterate方法，其返回的也是一个无限长度的Stream，与generate方法不同的是，其是通过函数f迭代对给指定的元素种子而产生无限连续有序Stream，其中包含的元素可以认为是：seed，f(seed),f(f(seed))无限循环。

Stream.iterate(T seed, UnaryOperator<T> f) 

示例：

Stream.iterate(1, item -> item + 1)
        .limit(10)
        .forEach(System.out::println); 
        // 打印结果：1，2，3，4，5，6，7，8，9，10 

4）empty() 方法

empty方法返回一个空的顺序Stream，该Stream里面不包含元素项。

（3）其他方法

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

2、Stream 流的操作

当把一个数据结构包装成 Stream 后，就要开始对里面的元素进行各类操作了。常见的操作可以归类如下：

（1）Intermediate 操作

1）map/flatMap

filter方法对原Stream按照指定条件过滤，在新建的Stream中，只包含满足条件的元素，将不满足条件的元素过滤掉。

2）filter

filter方法对原Stream按照指定条件过滤，在新建的Stream中，只包含满足条件的元素，将不满足条件的元素过滤掉。

3）sorted

对Stream的排序通过sorted进行，它比数组的排序更强之处在于你可以首先对Stream进行各类map、filter、limit、skip甚至distinct来减少元素数量后再排序，这能帮助程序明显缩短执行时间。

4）distinct

distinct 方法以达到去除掉原 Stream 中重复的元素，生成的新 Stream 中没有没有重复的元素

5）limit/skip

limit返回Stream的前面n个元素；skip则是扔掉前n个元素（它是由一个叫 subStream的方法改名而来）。

（2）Terminal 操作

1）forEach

forEach方法接收一个Lambda表达式，然后在Stream的每一个元素上执行该表达式。
注：forEach 不能修改自己包含的本地变量值，也不能用break/return之类的关键字提前结束循环。

2）findFirst

这是一个termimal兼short-circuiting操作，它总是返回Stream的第一个元素或者空。

3）reduce

这个方法的主要作用是把Stream元素组合起来。它提供一个起始值（种子），然后依照运算规则（BinaryOperator），和前面Stream的第一个、第二个、第n个元素组合。从这个意义上说，字符串拼接、数值的 sum、min、max、average都是特殊的reduce。

4）min/max

min和max的功能也可以通过对Stream元素先排序，再findFirst来实现，但前者的性能会更好为O(n)，而sorted的成本是O(nlogn)。同时它们作为特殊的reduce方法被独立出来也是因为求最大最小值是很常见的操作。

5）Match

Stream有三个match方法，从语义上说：
(1).allMatch：Stream 中全部元素符合传入的 predicate，返回 true；
(2).anyMatch：Stream 中只要有一个元素符合传入的 predicate，返回 true；
(3).noneMatch：Stream 中没有一个元素符合传入的 predicate，返回 true；
注：它们都不是要遍历全部元素才能返回结果。例如allMatch只要一个元素不满足条件，就skip剩下的所有元素，返回false。

6）limit

limit方法将截取原Stream，截取后Stream的最大长度不能超过指定值N。如果原Stream的元素个数大于N，将截取原Stream的前N个元素；如果原Stream的元素个数小于或等于N，将截取原Stream中的所有元素。

（3）总结

1）可以使用 filter、distinct、skip 和 limit 对流进行筛选和切片；

2）可以使用 map 和 flatMap 提取或转换流中的元素；

3）可以使用 findFirst 和 findAny 方法查找流中的元素，你可以用 allMatch、noneMatch 和 anyMatch 方法让流匹配给定的谓语；

4）上述方法都利用了短路：找到结果就立刻停止计算，并没有必要处理整个流；

5）可以利用 reduce 方法将流中的所有元素迭代合并成一个结果，例如求和或者查询最大的元素；

6）filter 和 map 等操作都是无状态的，他们并没有储存任何状态。reduce 等操作要储存状态才能计算出一个值，sorted 和 distinct 等操作也要储存状态，因为他们需要把流中的所有元素缓存起来才能返回一个新的流，这种操作称为有状态操作.

7）流有三种基本原始类型特化：intStream、doubleStream 和 LongStream，他们的操作也有相应的特化

8）流不仅可以从集合创建，也可以从值、数组、文件以及 iterate 与 generate 等特定方法创建。

四、小结

Stream 的特性可以归纳为：

1、不是数据结构；

2、它没有内部存储，它只是用操作管道从source（数据结构、数组、generator function、IO channel）抓取数据；

3、它也绝不修改自己所封装的底层数据结构的数据。例如 Stream 的 filter 操作会产生一个不包含被过滤元素的新 Stream，而不是从 source 删除那些元素；

4、所有 Stream 的操作必须以 lambda 表达式为参数；

5、不支持索引访问；

6、你可以请求第一个元素，但无法请求第二个，第三个，或最后一个；

7、很容易生成数组或者List;

8、惰性化；

9、很多Stream操作是向后延迟的，一直到它弄清楚了最后需要多少数据才会开始；

10、Intermediate操作永远是惰性化的；

11、并行能力；

12、当一个 Stream 是并行化的，就不需要再写多线程代码，所有对它的操作会自动并行进行的；

13、可以是无限的。集合有固定大小，Stream 则不必。limit(n)和findFirst()这类的short-circuiting操作可以对无限的Stream进行运算并很快完成。