WebFlux前置知识（三）

[TOC]

Stream 流式 API 也是 JDK8 开始引入的 API，用过的小伙伴都说爽，没用过的小伙伴则感到一脸懵逼。现在 JDK 已经到 16 了，可能很多人还不熟悉 Stream 流操作，这次趁着讲 WebFlux，松哥也来和大家一起回顾一下 Stream 流操作。

老实说，松哥在日常工作中，有时候刚好碰上了也会用 Stream 流，用着确实很爽，代码简洁，而且看起来高大上，确实是一个不错的东西。不过组里有很多人其实并不太熟悉 Stream 操作，所以也就没有去强行推广，但是作为一个合格的程序员，还是很有必要去学习一下 Stream 流操作的。

1.Stream 流简介

说到 Stream，大家可能很容易想到 Java IO 中的各种流操作，名字看起来很像，但这其实是两个完全不同的东西。

Java8 中的 Stream 不存储数据，它通过函数式编程模式来对集合进行链状流式操作。

2.基本玩法

先来个简单的例子感受下 Stream 的操作。

例如我有一个数组，如下：

int[] arr = {1, 2, 3, 4, 5, 6};

现在想给这个数组求平均数，如下：

double asDouble = IntStream.of(arr).average().getAsDouble();

一行代码搞定，首先调用 IntStream.of 方法获取流对象，然后调用 average 方法计算平均数，再调用 getAsDouble 方法获取结果。全程都是方法调用，不用自己写求平均数的逻辑。

再比如求和，如下：

int sum = IntStream.of(arr).sum();

也是直接调用 sum 方法即可。

而且我们还可以在流中对数据进行二次加工，例如给数组中的每个元素先求平方再求和：

double[] arr = {1, 2, 3, 4, 5, 6};
double sum = DoubleStream.of(arr).map(i -> Math.pow(i, 2)).sum();

先在 Map 中对每一个元素求平方，然后再求和。

这里就能涉及到两个概念：

• 中间操作
• 终止操作

所谓中间操作，就是中途求平方的操作，所谓终止操作就是最终计算出结果的操作，只要方法的返回值不是一个 Stream，那就是终止操作，否则就是中间操作。

3.Stream 的创建

获取一个流的方式多种多样，最常见的就是从集合或者数组中获取一个流，当然我们也可以直接自己创建一个流出来。

一起来看下。

集合

List 集合或者 Set 集合都可以直接搞一个流出来，方式如下：

List<String> list = new ArrayList<>();
Stream<String> s1 = list.stream();
Set<String> set = new HashSet<>();
Stream<String> s2 = set.stream();

集合中直接调用 stream 方法就可以获取到流。

数组

通过数组获取流的方式也很简单，如下：

IntStream stream = Arrays.stream(new int[]{11, 22, 33, 44, 55, 66});

数字 Stream

也可以直接利用 IntStream、LongStream 等对象创建一个数字 Stream，如下：

IntStream s1 = IntStream.of(1, 2, 3);
DoubleStream s2 = DoubleStream.of(1, 2, 3);
LongStream s3 = LongStream.of(1L, 2L, 3L);

自己创建

Random random = new Random();
Supplier<Integer> supplier = () -> random.nextInt(100);
Stream<Integer> stream = Stream.generate(supplier).limit(5);

调用 Stream.generate 方法可以自己创建一个流，自己创建的时候需要提供一个 Supplier，通过调用 Supplier 中的 get 方法自动获取到元素。

无论哪种创建方式，大家需要明白的是，Stream 并不会保存数据，它只会对数据进行加工。

4.Stream 的中间操作

中间操作可以分为两大类：

• map 或者 filter 会从输入流中获取每一个元素，并且在输出流中得到一个结果，这些操作没有内部状态，称为无状态操作。
• reduce、sum、max 这些操作都需要内部状态来累计计算结果，所以称为有状态操作。

分别来看下：

无状态操作

• map/mapToXxx
• flatMap/flatMapToXxx
• filter
• peek

有状态操作

• distinct
• sorted
• limit/skip

map

Stream.map() 是 Stream 中最常用的一个转换方法，可以把一个 Stream 对象转为另外一个 Stream 对象。map 方法所接收的参数就是一个 Function 对象，松哥在前面文章中和大家介绍过 Function 对象了，就是有输入有输出（参见WebFlux 前置知识（一） (opens new window)），了解了 map 的参数，那么 map 的功能就很明白了，就是对数据进行二次加工。

举个栗子，例如把一个字符串数组转为数字：

String[] arr = {"1", "2", "3"};
Stream<String> s1 = Arrays.stream(arr);
Stream<Integer> s2 = s1.map(i -> Integer.valueOf(i));

再比如一个数字流，给所有的元素乘 2，如下：

IntStream.of(1, 2, 3).map(i -> 2 * i).forEach(System.out::println);

最后的 forEach 就是将元素打印出来。

JDK 中也提供了一些现成的格式转换，如下图：

这样可以直接将元素转为 Double、Long、Obj 等类型，例如下面这样：

String[] arr = {"1", "2", "3"};
Stream<String> s1 = Arrays.stream(arr);
s1.mapToLong(i -> Long.parseLong(i)).forEach(System.out::println);

flatMap

flatMap 可以把 Stream 中的每个元素都映射为一个 Stream，然后再把这多个 Stream 合并为一个 Stream。

例如如下代码，返回的 Stream 中的元素是数组：

Stream<Integer[]> s = Stream.of(new Integer[]{1, 2, 3}, new Integer[]{4, 5, 6}, new Integer[]{7, 8, 9});
s.forEach(System.out::println);

通过 flatMap 我们可以将 Stream 中的元素变为 Integer：

Stream<Integer> s = Stream.of(new Integer[]{1, 2, 3}, new Integer[]{4, 5, 6}, new Integer[]{7, 8, 9}).flatMap( i -> Arrays.stream(i));
s.forEach(System.out::println);

filter

filter 操作会对一个 Stream 中的所有元素一一进行判断，不满足条件的就被过滤掉了，剩下的满足条件的元素就构成了一个新的 Stream。

例如要找到数组中所有大于 3 的元素，如下：

IntStream.of(2, 3, 4, 5, 6, 7).filter(i -> i > 3).forEach(System.out::println);

filter 方法接收的参数是 Predicate 接口函数，关于 Predicate 接口函数，大家可以参考WebFlux 前置知识（一） (opens new window)）一文。

peek

peek 的入参是 Consumer，没有返回值，因此当我们要对元素内部进行处理时，使用 peek 是比较合适的，这个时候可以不用 map（map 的入参是 Function，它是有返回值的）。peek 方法本身会继续返回流，可以对数据继续进行处理。

举个简单的数据转换的例子吧（最终返回的数据并不会被转换）：

IntStream.of(2, 3, 4, 5, 6, 7).filter(i -> i > 3).peek(String::valueOf).forEach(i-> System.out.println(i));

peek 方法的感觉就像数据中途被消费了一次。

distinct

这个是去重。由于去重操作需要获取到其他元素的值（比较之后才知道是否重复），所以这个是有状态操作。如下：

IntStream.of(2, 3, 4, 3, 7, 6, 2, 5, 6, 7).distinct().forEach(System.out::println);

sorted

sorted 是排序，因为也需要知道其他元素的值，然后才能去重，所以这个也是有状态操作，如下：

IntStream.of(2, 3, 4, 3, 7, 6, 2, 5, 6, 7).distinct().sorted().forEach(System.out::println);

limit/skip

limit 和 skip 配合操作有点像数据库中的分页，skip 表示跳过 n 个元素，limit 表示取出 n 个元素。例如下面这个例子：

Arrays.asList('A', 'B', 'C', 'D', 'E', 'F').stream().skip(2).limit(3).forEach(System.out::println);

这个会跳过 A 和 B，最终打印出 C D E。这也是一种有状态操作。

5.Stream 终止操作

终止操作就是最终计算出结果的操作，只要方法的返回值不是一个 Stream，那就是终止操作，否则就是中间操作。

终止操作又分为两类：

• 短路操作：不用处理全部元素就可以返回结果。
• 非短路操作：必须处理所有元素才能得到最终结果。

各自都包含哪些操作，我们分别来看下：

非短路操作

• forEach/forEachOrdered
• collect/toArray
• reduce
• min/max/count

短路操作

• findFirst/findAny
• allMatch/anyMatch/noneMatch

forEach/forEachOrdered

forEach 和 forEachOrdered 都是接收一个 Consumer 类型的参数，完成对参数的消费，不同的是，在并行流中，forEachOrdered 会保证执行顺序。

例如如下一段代码：

int[] arr = {1, 2, 3, 4, 5, 6, 7, 8, 9};
Arrays.stream(arr).parallel().forEach(System.out::println);
Arrays.stream(arr).parallel().forEachOrdered(System.out::println);

前者打印出来的顺序不一定是 123456789，后者一定是。

collect/toArray

这两个都是收集器，可以将执行结果转为一个 List 集合或者一个数组：

List<Integer> list = Stream.of(1, 2, 3, 4).filter(p -> p > 2).collect(Collectors.toList());
System.out.println(list);

reduce

reduce 是 Stream 的一个聚合方法，它可以把一个 Stream 的所有元素按照聚合函数聚合成一个结果。reduce 方法传入的对象是BinaryOperator 接口，它定义了一个apply 方法，负责把上次累加的结果和本次的元素进行运算，并返回累加的结果。

举个简单的例子，数组求和，当然可以直接调用 sum 计算，我们这里也可以调用 reduce 来实现，如下：

Optional<Integer> optional = Stream.of(1, 2, 3, 4).reduce((i, j) -> i + j);
System.out.println(optional.orElse(-1));

reduce 的参数是 BinaryOperator，这个接收两个参数，第一个参数是之前计算的结果，第二个参数是本次参与计算的元素，两者累加求和。

再比如给一段话中间加上.，方式如下：

Optional<String> s = Stream.of("wwwjavaboyorg".split("")).reduce((i, j) -> i + "." + j);
System.out.println(s.orElse(""));

最终执行结果如下：

w.w.w.j.a.v.a.b.o.y.o.r.g

min/max/count

这个就比较简单了，就是求最大值最小值，统计总个数，如下表示统计总个数：

Stream<Integer> s = Stream.of(1, 2, 3, 4);
long count = s.count();
System.out.println("count = " + count);

如下表示统计最小值：

Stream<Integer> s = Stream.of(1, 2, 3, 4);
Optional<Integer> min = s.min(Comparator.comparingInt(i -> i));
System.out.println("min.get() = " + min.get());

findFirst/findAny

这两个就是返回流中的第一个、任意一个元素，findAny 要在并行流中测试才有效果，举个栗子：

for (int i = 0; i < 10; i++) {
    Optional<Integer> first = Stream.of(1, 2, 3, 4).parallel().findFirst();
    System.out.println("first.get() = " + first.get());
}
System.out.println("=============");
for (int i = 0; i < 10; i++) {
    Optional<Integer> first = Stream.of(1, 2, 3, 4).parallel().findAny();
    System.out.println("first.get() = " + first.get());
}

allMatch/anyMatch/noneMatch

allMatch、anyMatch、noneMatch 用来判断所有元素、任意元素或者没有元素满足给定的条件。这三个方法的参数都是一个 Predicate 接口函数。

boolean b = Stream.of(1, 2, 3, 4).allMatch(i -> i > 5);
System.out.println("b = " + b);

6.并行流

通常情况下，对 Stream 的元素进行处理是单线程的，即一个一个元素进行处理。有时候我们希望可以并行处理 Stream 元素，因为在元素数量非常大的情况，并行处理可以大大加快处理速度。

把一个普通 Stream 转换为可以并行处理的 Stream 非常简单，只需要用 parallel 方法进行转换：

new Random().ints().limit(50).parallel().forEach(i->{
    System.out.println(Thread.currentThread().getName() + "--->" + i);
});

这样数据在后台就是并行打印的。

7.收集器

收集器可以将计算结果重新整理收集到一个集合中，这个集合可以是一个 List/Set 获取其他，并且还可以在收集的过程中对数据进行处理。

例如我有一个 users 集合，里边保存了用户数据，用户有 username、age 以及 gender 三个属性，如下代码分别表示：

• 提取出用户对象中的 age 属性组成新的集合并返回。
• 提取出用户对象中的 username 属性组成新的集合并返回。
• 提取出用户对象中的 gender 属性组成新的集合并返回（这里是一个 Set 集合，所以会自动去重）。

List<Integer> ages = users.stream().map(User::getAge).collect(Collectors.toList());
System.out.println("ages = " + ages);
List<String> usernames = users.stream().map(User::getUsername).collect(Collectors.toList());
System.out.println("usernames = " + usernames);
Set<String> genders = users.stream().map(User::getGender).collect(Collectors.toSet());
System.out.println("genders = " + genders);

Collectors.toList() 最终返回的是 ArrayList，Collectors.toSet() 最终返回的是 HashSet。

如果我们想返回一个 Vector 或者 TreeSet，也是可以的，如下：

List<Integer> ages = users.stream().map(User::getAge).collect(Collectors.toList());
System.out.println("ages = " + ages);
List<String> usernames = users.stream().map(User::getUsername).collect(Collectors.toCollection(Vector::new));
System.out.println("usernames = " + usernames);
TreeSet<String> genders = users.stream().map(User::getGender).collect(Collectors.toCollection(TreeSet::new));
System.out.println("genders = " + genders);

也可以获取某一个字段的统计信息：

IntSummaryStatistics ageStatistics = users.stream().collect(Collectors.summarizingInt(User::getAge));
System.out.println("ageStatistics = " + ageStatistics);

这个统计信息中包含：总和、最小值、平均值以及最大值等：

ageStatistics = IntSummaryStatistics{count=20, sum=1222, min=9, average=61.100000, max=96}

还可以对数据进行分块，将男女不同性别统计出来：

Map<Boolean, List<User>> map = users.stream().collect(Collectors.partitioningBy(u -> u.getGender().equals("男")));
System.out.println("map = " + map);

也可以按照性别对数据进行分组，如下：

Map<String, List<User>> map2 = users.stream().collect(Collectors.groupingBy(User::getGender));
System.out.println("map2 = " + map2);

分组后，Map 中的 key 就是性别；分块后，Map 中的 key 就是 true/false。

再比如统计男女的人数：

Map<String, Long> map2 = users.stream().collect(Collectors.groupingBy(User::getGender,Collectors.counting()));
System.out.println("map2 = " + map2);

好啦，今天我们就先聊这么多～