Java8 Stream 入门

系统的整理一下 java8 Streams 的使用。

思想

是函数式编程（functional programming）的一种 Java 实现

强调将计算过程分解成可复用的函数，主要使用 map 方法和 reduce 方法组合而成的 MapReduce 算法，最好的实现 Apache Hadoop

关于函数式编程，请参考阮一峰 的 函数式编程初探

Streams 和 Collections 的不同

• 不储存元素。 Stream 不是储存元素的数据结构；相反的，它通过管道对源就像数据结构、数组、构造方法、IO流的元素进行操作。
• 纯粹的方法。 Stream 上的操作会产生结果，但不会修改其来源。例如，过滤从集合获取的流会生成一个没有过滤元素的新 Stream，而不是从源集合中删除元素。
• 惰性化。 许多流操作（例如过滤，映射或重复移除）可以被懒性化实现，从而为优化提供机会。例如，“find the first String with three consecutive vowels”不需要检查所有的输入字符串。流操作分为中间 intermediate (Stream-producing) 操作和终端 terminal (value-or side-effect-producing) 操作，intermediate 操作总是惰性的。
• 可能没有限制。 尽管集合的大小有限，但流不需要。诸如 limit(n) 或 findFirstf() 之类的短路操作可以允许无限流上的计算在有限的时间内完成。
• 一次性。 流的元素在流的生命周期中仅访问过一次。像 Iterator 一样，必须生成一个新的流来重新访问源的相同元素。

集合和流，它们有不同的关注点，集合主要关注集合的有效管理和访问。
相反，流不直接提供访问和操作元素的手段，而是关注于声明性地描述它们的来源和将在该来源上进行的计算操作。如果流操作没有你想要的功能，你可以使用 iterator() 或 spliterator() 来遍历操作。
**
增强 for 循环内部还是使用 iterator() 来进行遍历，它们同属于外部遍历器，java8 集合的 forEach 和 stream 的 forEach() 属于内部遍历器，流的内部遍历器可以使用到流的并行（parallel）特性，从而加快速度。
**

• 枚举，迭代器和增强的 for 循环都是外部迭代器（记着方法 iterator()，next() 或 hasNext() 吗？）。
• java8 集合 forEach 和 stream 的 forEach 为外部迭代器。

---

流操作分为中间操作（intermediate operations）和终端操作（terminal operations），结合形成流管道。流管道由源（例如集合，数组，生成器函数或 I/O 通道）组成; 随后是零个或多个中间操作，例如 Stream.filter 或 Stream.map 和诸如 Stream.forEach 或 Stream.reduce 之类的终端操作。

Intermediate operations（中间操作）

中间操作返回一个新的流（Stream<T>）。

他们总是惰性的，执行诸如 filter() 之类的中间操作实际上并不执行任何过滤，而是创建一个新的流，该流在遍历时包含与给定谓词相匹配的初始流的元素。在管道的终端操作被执行时对源的流水遍历才会开始。

中间操作进一步分为无状态和有状态操作。无状态操作（如 filter 和 map）在处理新元素时不会保留先前看到的元素的状态 -- 每个元素可以独立于其他元素上的操作进行处理。有状态的操作（如 distinct 和 sorted）可能会在处理新元素时结合之前看到的元素的状态。

有状态的操作可能需要在生成结果之前处理整个输入。例如，只有在查看了流的所有元素之后，才能对排序流产生任何结果。因此，在并行计算中，一些包含有状态中间操作的管道可能需要对数据进行多次传递，或者可能需要缓存重要数据。只包含无状态中间操作的流水线可以一次处理，无论是顺序处理还是并行处理，只需最少的数据缓冲。

下面列举个人常用的一些操作：

1. map

返回由给定函数作用于此流的元素后产生的结果组成的流。

给定函数为无干涉，无状态的操作作用于每个元素。不然之后的操作结果可能不会很准确。

• 无干涉

  无干涉主要是指在流操作期间不去修改源流。

• 无状态

  无状态是指我们在处理时不产生中间状态，操作不依赖之前的状态。

2. distinct

distinct 保证输出的流中包含唯一的元素，它是通过 Object.equals(Object) 来检查是否包含相同的元素。它是一个有状态的中间操作。

在并行流中对无序数组去重效率更高，对于有序数组可以使用 unordered() 无序检索提高速度，或者使用sequential()来实现串行。

相反有序数组更适合使用串行流。

3. peek

peek 产生一个和原流相同的流，并在遍历流的过程中去消费每个元素。

使用 peek 的主要目的是“看，不要动”
此方法主要用于支持调试，您希望在元素流经管道中的某个点时看到这些元素：请谨慎使用此方法作为副作用，因为它有可能会修改源流。

4. flatMap

返回一个流，该流包含将原流的每个元素替换为映射函数应用于每个元素而生成的映射流的内容的结果。每个映射流都将其内容放入此流后关闭。（如果映射流为空，则使用空流代替）
简而言之，就是将原流每个元素通过映射函数生成的新流组合成一个新流。

flatMap() 操作具有对流的元素应用一对多转换，然后将生成的元素展平为新流的效果。

example

orders 是采购订单流，并且每个采购订单都包含一系列采购列，则以下内容会生成包含所有订单中的所有采购列的流：

orders.flatMap(order -> order.getLineItems().stream())...

如果 path 是文件的路径，那么下面的内容会生成包含在该文件中的单词流：

    Stream<String> lines = Files.lines(path, StandardCharsets.UTF_8);
    Stream<String> words = lines.flatMap(line -> Stream.of(line.split(" +")));

上面 flatmap 中的映射函数使用的正则比较简单，具体单词划分的正则不是这样。

5. count

返回此流中元素的数量。这是一个简写，相当于：

    return mapToLong(e -> 1L).sum();

Terminal operations（终端操作）

终端操作返回确定类型的结果

如 Stream.forEach 或 IntStream.sum，可能会遍历流以产生结果或副作用(side-effect)。终端操作执行后，流管道被视为消耗，并不能再使用；如果你需要再次遍历相同的数据源，则必须返回到数据源以获取新的流。在几乎所有情况下，终端操作都非常急切，在返回之前完成数据源的遍历和管道的处理。只有终端操作 iterator() 和 spliterator() 不是。

副作用（side-effect）

副作用可能会违反无状态要求和对线程安全产生危害。
许多计算可能会产生副作用，但是可以更安全有效地表达，而不会产生副作用，例如使用 reduction 而不是 mutable accumulators。少量流操作（例如 forEach() 和 peek()）只能通过副作用操作；这些应该小心使用。
比如我们在对流操做以期望得到想要的结果，而无意修改了原始流，便产生了副作用。

折叠（Reduction operations）

归约操作（也称为折叠）采用一系列输入元素，并通过重复应用组合操作（例如查找一组数字的和或最大值）或将元素累加到列表中来将它们组合为单个汇总结果。流类具有多种形式的通用归约操作，称为 reduce() 和 collect()，以及多个专用简化形式，如 sum()，max() 或 count()。

可变归约（Mutable reduction）

可变归约操作将输入元素累加到可变结果容器中，例如 Collection 或 StringBuilder，因为它处理流中的元素。

可变缩减操作称为 collect()，因为它将所需结果一起收集到结果容器（如集合）中。 收集操作需要三个功能：构造结果容器的新实例的供应者函数，将输入元素并入结果容器的累加器函数以及将一个结果容器的内容合并到另一个结果容器的组合函数。

• 供应器 (supplier())

• 累加器 (accumulator())

  • 组合器 (combiner())
  • 修整器 (finisher()) 可省略

<R> R collect(Supplier<R> supplier,
               BiConsumer<R, ? super T> accumulator,
               BiConsumer<R, R> combiner);

例如下面的代码：

ArrayList<String> strings = new ArrayList<>();
for (T element : stream) {
   strings.add(element.toString());
} 

我们可以写成：

ArrayList<String> strings = stream.collect(() -> new ArrayList<>(),
                                          (c, e) -> c.add(e.toString()),
                                          (c1, c2) -> c1.addAll(c2));

简写作：

List<String> strings = stream.map(Object::toString)
                                  .collect(ArrayList::new, ArrayList::add, ArrayList::addAll);

在这里，我们的供应器是 ArrayList 构造函数，累加器将字符串化的元素添加到 ArrayList，组合器只是简单地使用 addAll 将字符串从一个容器复制到另一个容器中。
collect 的供应器，累加器和组合器三个方面紧密耦合。我们可以使用抽象的 Collector 来包含三个方面，上面的代码可以重写为：

List<String> strings = stream.map(Object::toString)
                                  .collect(Collectors.toList());

收集器（Collectors）实现类

一种可变减少操作，将输入元素累加到可变结果容器中，可选地，在处理完所有输入元素后，将累加结果转换为最终表示形式。缩减操作可以按顺序执行也可以并行执行。
可变减少操作的例子包括：

• 将元素累加到集合中；
  toList toMap toSet toCollection
• 使用 StringBuilder 连接字符串；
  joining
• 计算关于总和，最小值，最大值或平均值等元素的摘要信息；
  • 求和 counting() collectingAndThen
  • 汇总 summarizingDouble summingDouble...
  • 最大值、最小值 maxBy minBy
  • 平均值 averagingDouble averagingInt...
• 计算“数据透视表”摘要，例如“卖方最大价值交易”等。
  • 分组 groupingBy
  • 分割 partitioningBy
    例子：

// Accumulate names into a List
List<String> list = people.stream().map(Person::getName).collect(Collectors.toList());
     
// Accumulate names into a TreeSet
Set<String> set = people.stream().map(Person::getName).collect(Collectors
                                                      .toCollection(TreeSet::new));

// Convert elements to strings and concatenate them, separated by commas
String joined = things.stream()
                         .map(Object::toString)
                         .collect(Collectors.joining(", "));

// Compute sum of salaries of employee
int total = employees.stream()
                      .collect(Collectors.summingInt(Employee::getSalary)));

// Group employees by department
Map<Department, List<Employee>> byDept = employees.stream()
                    .collect(Collectors.groupingBy(Employee::getDepartment));

// Compute sum of salaries by department
Map<Department, Integer> totalByDept = employees.stream()
                    .collect(Collectors.groupingBy(Employee::getDepartment,
                                         Collectors.summingInt(Employee::getSalary)));

// Partition students into passing and failing
Map<Boolean, List<Student>> passingFailing = students.stream()
             .collect(Collectors.partitioningBy(s -> s.getGrade() >= PASS_THRESHOLD));