Java-8-流

发表于 更新于 分类于 ， Valine：

Java 8 中引入的 新的类 Stream，提供了一种对于 Java 集合类的更高层次的概念抽象。流可以自动处理很多计算、操作过程中的琐碎细节，使得代码更加高效简洁。

A. 流的基本操作

我们使用一个例子来展示使用 for循环 和 流 来遍历处理集合元素的区别。

1
2
3
4
5
6
7
8
9
10
11
12
List<String> words = List.of(contents.split("\\PL+"));  // 待处理的对象
// 使用迭代的方式处理- 计算长度达于12的字符串元素的个数
int count = 0;
for(String w : words){
    if(w.length() > 12) count++;
}

// 使用 stream 来处理
long count = words.stream() 
    .filter(w -> w.length() > 12)
    .count();
//stream 可以替换为 parallelStream，这样可以并行处理 filter 与 count

上面的两组程序实现的是相同的功能，可以发现，对于使用 Stream 处理的方案，程序中只体现了要做什么(what)，但具体的实现细节（How）基本都被省略了。

我们进一步分析 使用 stream 的程序，它可以分为下面三个步骤：

  1. 制造一个流；（words.stream()）
  2. 指定 中间操作（intermediate operation），来处理最初的输入流；（.filter(), 从原始的流中取得一个新的流，with流中的单词长度大于12）
  3. 执行 最终操作（terminal operation）来生成结果（count() 方法，根据中间步骤新生成的流来获取结果）。

流与集合有很多相似的地方 (e.g. 都允许我们对数据进行变换，操作)。但他们有如下几点重要的区别需要注意：

  1. 一个 stream 不存储元素，元素仍然存储在集合类中；
  2. 流操作不会改变他们的源，例如上面的 filter 方法，不会改变 源 stream 中的元素，会生成一个新的 stream。
  3. stream 的操作是一种 lazy 操作，也就是说，他们只会在结果被需要的时候执行。例如上面的程序中，如果只要求返回前五个长字符串，filter 方法会在得到第五个值后停下。也因此，stream 的方法可以处理 infinite 的 stream。

a. 生成 Stream 的方法

生成 Stream 的方法主要可以分为下面几种：

  1. 生成空的 stream：
    1. 使用 Stream.empty();。
  2. 生成无穷个元素的 stream：
    1. 使用 Stream.generate(),或者使用 Stream.iterate() 使用有规律的元素填充 stream（e.g.等差数列）。
  3. 生成有限个元素的 stream：
    1. 从集合生成 stream，使用 集合类的实例方法 .stream()；
    2. 从数组生成 stream，使用 Stream.of(数组量String[])，或者使用 Arrays.stream()；
    3. 使用 Stream.iterate() 生成，并规定 迭代停止的条件。

b. Stream 的常用变换方法

filter(), map()

上面的例子中，我们使用 filter 将原始的流进行过滤，得到新的流，filter() 中需要传入的是一个 Predicate<T> 接口（接口中的函数输入T返回 boolean值）。

map() 也是一种转换方法，它将原流中的元素通过映射规则生成新的流，在下面的例子中，map 方法将原流中的字母全部转换为了小写。

1
2
// words 是一个 List<String>
Stream<String> lowercaseWords = words.stream().map(String::toLowerCase);

flatMap() 方法可以将所有元素以一个 stream 的形式输出，例如 将 Stream<stream<String>> 变为 Stream<String>类型。

截取、拼接流

截取一个 流 有如下方法：

  1. limit(n) , 返回前n个元素构成新的流；skip(n), 跳过前面n个元素构成新的流；
  2. takeWhile(), 取用原流中所有满足括号中条件的元素；dropWhile(), 当括号中条件满足是，舍弃对应元素；

拼接流： 使用静态方法 Stream.concat() 拼接两个流。

其他方法

distinct() 方法：检视原流，将原流中重复的元素丢弃后 按照原流中的顺序返回元素。

sorted() 方法：将流中的元素排序，需要元素实现 Comparable 接口，或者传入 Comparator 对象。

peak() 方法，用于 调试，原流中的元素不变，但当元素被取用时，调用 peak 中的方法。

1
2
3
4
5
Object[] powers = Stream.iterate(1.0, p -> p * 2)
   .peak(e -> System.out.println("Fetching " + e))
   .limit(20).toArray();
   // powers 为 iterate 生成的对象序列的前20个值，
   // peak 方法在每个值被取到的时候打印 "Fetching " + 值

c. 终端方法

最终，我们需要将 Stream 转换为供程序使用的数据类型，在上面的例子中，count 就是这样的作用-统计流中的元素的个数并返回。这种函数又叫做 reductions，可以将流 reduce 为其他类型的对象。

这里，简单介绍常用的 reduction 方法：

  1. count(), 返回流中元素的个数；
  2. max(), min(), 返回流中 最大最小值，需要注意的是，返回的值为 Optional<T> 类型，这样的好处是可以将 结果为 null 的情况也统一起来，都用 Optional<T> 类型的数据进行返回。
  3. findFirst(), 可以返回流中的第一个元素的值。同样返回 Optional<T> 类型。findAny() 返回流中所有元素的值，返回 Optional<T> 类型，此外，findAny() 方法在 parallel 的情况下效率很高。（参考下面例子）
  4. allMatch(), noneMatch(), anyMatch() 判断流中是否有匹配的元素，返回 boolean 类型的值，同样在 parallel 的情况下效率高。
  5. Part C 中介绍的将流存储在 数组、集合类、字符串中，也属于终端方法。
1
2
3
Optional<String> startsWithQ = words.parallel()
   .filter(s -> s.startsWith("Q"));
// 使用 parallel 使得程序可以 parallel 运行，提升效率

特别地，reduce() 方法可以根据 stream 中的元素计算得到一个值。例子如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
// 依次累加流中的元素，最后返回流中元素的和
List<Integer> values = ...;
Optional<Integer> sum = values.stream().reduce((x,y) -> x + y);

// 提供一个初始值（加法单位元），从而可以保证返回值不为空，从而不用处理 Optional<T> 类型对象
Integer sum = values.stream().reduce(0, (x,y) -> x + y);

// 求解字符串流中字符串的长度总和
// 适用于并行程序的写法
// words 是一个字符串的流
int res = words.reduce(0,
   (total, word) -> total + word.length(),
   (total1 + total2) -> total1 + total2);

最后一个适用于 并行程序的程序中，reduce() 函数中有三个参数，第二个参数为求字符串长度，第三个参数将不同线程中的计算结果相加。

B. Optional 类

Optional<T> 对象是 T 类型元素 or null 的包装。它提供了一种更加安全的引用，用于指向一个可能为空的元素。

a. 使用场景

具体的， Optional<T> 对象有如下两种用法，使得赋值，传值时更加安全。

用法1：当 Optional<T> 为 null时，提供一个默认值用于赋值。

1
2
3
4
5
6
7
// Optional<T> 对象的 orElse 函数。
// 当 Optional<T> 不为空时，用其中的值赋值给 res，否则，赋值 “ ”
String res = optionalString.orElse(" ");
// orElseGet()
String res = optionalString.orElseGet(() -> System.getProperty("myapp.default"));
// 为空时抛出异常
String res = optionalString.orElseThrow(IllegalStateException::new);

用法2：利用 Optional<T> 对象的 ifPresent 方法，判断其中是否有元素，然后根据判断结果决定后面程序如何执行。看例子：

1
2
3
4
5
6
// 如果 optionalValue 中有值，则process v。
optionalValue.ifPresent(v -> Process v);
// 如果 optionalValue 中有值，执行前半句，否则，执行后半句
optionalValue.ifPresentElse(
   v -> process v,
   () -> Do sth);

b. 生成与转换

生成 Optional 类型

Optional.empty() 可以生成空的 Optional 对象，Optional.of(元素值)， 可以生成带元素的Optional 对象。

flatMap()

我们可以利用 faltMap() 函数进行 不同泛型类 Optional 对象的转换，以及 Optional 对象和 Stream 之间的转换，参考下面的例子：

1
2
3
4
5
6
7
8
9
10
11
12
/**
* f() 函数返回 Optional<T>, T 上有 g() 函数，返回 Optional<U>
* 使用 flatMap() 函数进行转换
*/
Optional<U> res = s.f().flatMap(T::g);

// Optional 对象和 stream 之间的转换（拆包）
// 设 Users 有 lookup 方法如下
Optional<User> lookup(String id){...}
// 返回 Stream<User> 类型对象, 调用 stream() 方法
Stream<User> users = ids.map(Users::lookup)
   .flatMap(Optional::stream);

C. 将流中元素存储到数组或集合类中

a. 遍历

stream 对象的 forEach 方法，可以在遍历 流中的元素的同时执行操作，例如stream.forEach(System.out::println);.

b. 存储到数组中

使用 stream 对象的 toArray() 方法，默认的，stream 的 toArray() 方法返回 Object[], 我们可以传入对应数组的构造函数来构建特定类型的数组，如下：

1
String[] res = stream.toArray(String[]::new);

c. 存储到集合类中

使用 stream 对象的 collect() 方法，然后在该方法中传入 Collectors 类的静态工厂方法，构建特定的集合类型。参考例子如下：

1
2
List<String> res = stream.collect(Collectors.toList());
Set<String> res = stream.collect(Collectors.toSet());

将流中的对象存储到 Map 中相对复杂一些，需要同时 给出键与值 的内容。例子如下：

1
2
3
4
// people 是一个 Stream<Person> 类型的对象
Map<Integer, Person> idToPerson = people.collect(Collectors.toMap(
   Person::getId, Function.identity()));
   // Function.identity(), 返回一个与输入相同的对象，这里就是Person 类的对象

需要注意的是，对于 Map，可能出现 key 重复的现象，从而引起 exception，可以通过向 toMap() 传入额外的参数解决这一问题（此略）。

如果想将元素分组存储到 Map 中，可以通过 Collectors 的 groupingBy() 和 partitioningBy方法（推荐predicate function使用，即返回 boolean 类型的function）。例子如下：

1
2
3
4
5
6
7
// 例1- 利用 groupingBy 将元素分组存放到 Map 中
Map<String List<Locale>> countryToLocales = locales.collect(
   Collectors.groupingBy(Locale::getCountry));

// 例2- 利用 groupingBy 额外的参数，设置Map中的值
Map<String Long> countryToLocaleCounts = locales.collect(
   Collectors.groupingBy(Locale::getCountry, counting()));

类似上面例2 中所示，我们还可以向 groupingBy() 方法中传入如 summingInt(), maxBy() 等方法，来配置 Map 中key对应的值。

d. 存储到字符串中

可以将流中的元素直接组成生成字符串，向 collect() 方法中 传入 Collectors 的 joining 方法。

1
2
// 将stream中的对象全部转为 String 类型后，再用 joining 方法拼接起来，中间以 ", " 间隔。
String res = stream.map(Object::toString).collect(Collectors.joining(", "));

D. 基本数据类型的流 与 并行流

a. 基本数据类型的流

类似 集合类，Stream 方法只接收 Object 类型元素，这对于基本数据类型来说，效率相对低下。因此，对于基本数据类型，我们推荐使用 IntStream (for: char, shrot, byte, int, boolean), LongStream, DoubleStream.

生成

生成基本数据类型的流大致有下面几种方案：

  1. 生成 Stream 的方法基本都可使用，
    1. 例如 Arrays.stream()(利用数组生成), 或使用静态方法 generate(), iterate()。
    2. 使用 类型对应的 of() 方法, e.g. IntStream.of(1, 1, 2);
  2. 使用 range(), rangeClosed() 方法，例如：IntStream.range(0, 100) 将 0 到 100 （不含100）间的整数放入流中。
  3. 一些方法返回 基本数据类型的流，例如 CharSequence 接口的 codePoints() 方法，返回 IntStream 包含所有元素的 unicode 值。例如：sentence 为 String， IntStream codes = sentence.codePoints();
  4. 由 Stream 对象通过调用 mapToInt() 等方法生成。

与 Stream 之间的转换

我们可以将 流转为 基本数据类型的流，使用 Stream 的实例方法 mapToInt(). 同样，也可以将基本数据类型的流转为 Stream，使用 基本数据类型 流的 boxed() 方法。

基本数据类型流与流的区别

如下：

  • toArray 方法返回不同，基本类型的流返回 基本数据类型的数组；
  • 基本数据类型的流 对于流中返回 Optional 对象的方法，返回 OptionalInt, OptionalLong, .. 这些类型与 Optional 类型的区别是 get 方法，他们的 get 方法为 getAsInt, getAsLong ..
  • 基本数据类型的流 有 sum, average, max 等直接处理流中数据的方法，而 Stream 类没有。
  • 基本数据类型的流 的 summaryStatistics 方法，可以同时返回流中数据的 和，均值，最大，最小值。

b. 并行流

Stream 让并行的操作变得简单，对于流来说，大部分的并行的工作都会自动的处理。

实现 Stream 的并行操作，首先是 生成 一个可并行处理的 Stream，方法有两个：

  • 直接从集合类生成：collection.parallelStream();
  • 对于一个非 parallel 的 Stream 对象，使用 .parallel() 方法使之变为 可 parallel 处理的stream。

对于 parallel 执行的流，我们需要注意必须保证 对流的操作 是不依赖于可能改变的状态的，或者说是不依赖于元素执行的顺序。（否则，可能发生错误，例如 race condition的问题）

通常对于不需要关注顺序的操作，parallel 的效率会更高，我们可以使用 unordered() 方法， 指明可以按任意顺序执行，例子如下：

1
2
Stream<String> sample = words.parallelStream().unordered().limit(n);
// 相当于从流中不按顺序任取 n 个值。