一、Stream 流是如何工作的?

Java8 Stream是函数式编程,可以利用一系列的API对字节流做流式的处理。
流表示包含着一系列元素的集合,我们可以对其做不同类型的操作,用来对这些元素执行计算。听上去可能有点拗口,让我们用代码说话:

  1. List<String> myList =
  2. Arrays.asList("a1", "a2", "b1", "c2", "c1");
  3. myList
  4. .stream() // 创建流
  5. .filter(s -> s.startsWith("c")) // 执行过滤,过滤出以 c 为前缀的字符串
  6. .map(String::toUpperCase) // 转换成大写
  7. .sorted() // 排序
  8. .forEach(System.out::println); // for 循环打印
  9. 输出:
  10. // C1
  11. // C2

我们可以对流进行中间操作或者终端操作。小伙伴们可能会疑问?什么是中间操作?什么又是终端操作?
**Java8 Stream - 图1

  • :中间操作会再次返回一个流,所以,我们可以链接多个中间操作,注意这里是不用加分号的。上图中的filter 过滤,map 对象转换,sorted 排序,就属于中间操作。
  • :终端操作是对流操作的一个结束动作,一般返回 void 或者一个非流的结果。上图中的 forEach循环 就是一个终止操作。

看完上面的操作,感觉是不是很像一个流水线式操作呢。
实际上,大部分流操作都支持 lambda 表达式作为参数,正确理解,应该说是接受一个函数式接口的实现作为参数。

二、不同类型的 Stream 流

我们可以从各种数据源中创建 Stream 流,其中以 Collection 集合最为常见。如 ListSet 均支持 stream() 方法来创建顺序流或者是并行流。

并行流是通过多线程的方式来执行的,它能够充分发挥多核 CPU 的优势来提升性能。本文在最后再来介绍并行流,我们先讨论顺序流:

  1. Arrays.asList("a1", "a2", "a3")
  2. .stream() // 创建流
  3. .findFirst() // 找到第一个元素
  4. .ifPresent(System.out::println); // 如果存在,即输出
  5. // a1

在集合上调用stream()方法会返回一个普通的 Stream 流。但是, 您大可不必刻意地创建一个集合,再通过集合来获取 Stream 流,您还可以通过如下这种方式:

  1. Stream.of("a1", "a2", "a3")
  2. .findFirst()
  3. .ifPresent(System.out::println); // a1

例如上面这样,我们可以通过 Stream.of() 从一堆对象中创建 Stream 流。

除了常规对象流之外,Java 8还附带了一些特殊类型的流,用于处理原始数据类型intlong以及double。说道这里,你可能已经猜到了它们就是IntStreamLongStream还有DoubleStream

其中,IntStreams.range()方法还可以被用来取代常规的 for 循环, 如下所示:

  1. IntStream.range(1, 4)
  2. .forEach(System.out::println); // 相当于 for (int i = 1; i < 4; i++) {}
  3. // 1
  4. // 2
  5. // 3

上面这些原始类型流的工作方式与常规对象流基本是一样的,但还是略微存在一些区别:

  • 原始类型流使用其独有的函数式接口,例如IntFunction代替FunctionIntPredicate代替Predicate
  • 原始类型流支持额外的终端聚合操作,sum()以及average(),如下所示:
  1. Arrays.stream(new int[] {1, 2, 3})
  2. .map(n -> 2 * n + 1) // 对数值中的每个对象执行 2*n + 1 操作
  3. .average() // 求平均值
  4. .ifPresent(System.out::println); // 如果值不为空,则输出
  5. // 5.0

但是,偶尔我们也有这种需求,需要将常规对象流转换为原始类型流,这个时候,中间操作 mapToInt()mapToLong() 以及mapToDouble就派上用场了:

  1. Stream.of("a1", "a2", "a3")
  2. .map(s -> s.substring(1)) // 对每个字符串元素从下标1位置开始截取
  3. .mapToInt(Integer::parseInt) // 转成 int 基础类型类型流
  4. .max() // 取最大值
  5. .ifPresent(System.out::println); // 不为空则输出
  6. // 3

如果说,您需要将原始类型流装换成对象流,您可以使用 mapToObj()来达到目的:

  1. IntStream.range(1, 4)
  2. .mapToObj(i -> "a" + i) // for 循环 1->4, 拼接前缀 a
  3. .forEach(System.out::println); // for 循环打印
  4. // a1
  5. // a2
  6. // a3

下面是一个组合示例,我们将双精度流首先转换成 int 类型流,然后再将其装换成对象流:

  1. Stream.of(1.0, 2.0, 3.0)
  2. .mapToInt(Double::intValue) // double 类型转 int
  3. .mapToObj(i -> "a" + i) // 对值拼接前缀 a
  4. .forEach(System.out::println); // for 循环打印
  5. // a1
  6. // a2
  7. // a3

三、Stream 流的处理顺序

上小节中,我们已经学会了如何创建不同类型的 Stream 流,接下来我们再深入了解下数据流的执行顺序。
在讨论处理顺序之前,您需要明确一点,那就是中间操作的有个重要特性 —— 延迟性。观察下面这个没有终端操作的示例代码:

  1. Stream.of("d2", "a2", "b1", "b3", "c")
  2. .filter(s -> {
  3. System.out.println("filter: " + s);
  4. return true;
  5. });

执行此代码段时,您可能会认为,将依次打印 “d2”, “a2”, “b1”, “b3”, “c” 元素。然而当你实际去执行的时候,它不会打印任何内容。

为什么呢?
**
原因是:当且仅当存在终端操作时,中间操作操作才会被执行。

是不是不信?接下来,对上面的代码添加 forEach终端操作:

  1. Stream.of("d2", "a2", "b1", "b3", "c")
  2. .filter(s -> {
  3. System.out.println("filter: " + s);
  4. return true;
  5. })
  6. .forEach(s -> System.out.println("forEach: " + s));

再次执行,我们会看到输出如下:

  1. filter: d2
  2. forEach: d2
  3. filter: a2
  4. forEach: a2
  5. filter: b1
  6. forEach: b1
  7. filter: b3
  8. forEach: b3
  9. filter: c
  10. forEach: c

输出的顺序可能会让你很惊讶!你脑海里肯定会想,应该是先将所有 filter 前缀的字符串打印出来,接着才会打印 forEach 前缀的字符串。

事实上,输出的结果却是随着链条垂直移动的。比如说,当 Stream 开始处理 d2 元素时,它实际上会在执行完 filter 操作后,再执行 forEach 操作,接着才会处理第二个元素。

是不是很神奇?为什么要设计成这样呢?
原因是出于性能的考虑。这样设计可以减少对每个元素的实际操作数,看完下面代码你就明白了:

  1. Stream.of("d2", "a2", "b1", "b3", "c")
  2. .map(s -> {
  3. System.out.println("map: " + s);
  4. return s.toUpperCase(); // 转大写
  5. })
  6. .anyMatch(s -> {
  7. System.out.println("anyMatch: " + s);
  8. return s.startsWith("A"); // 过滤出以 A 为前缀的元素
  9. });
  10. // map: d2
  11. // anyMatch: D2
  12. // map: a2
  13. // anyMatch: A2

终端操作 anyMatch()表示任何一个元素以 A 为前缀,返回为 true,就停止循环。所以它会从 d2 开始匹配,接着循环到 a2 的时候,返回为 true ,于是停止循环。

由于数据流的链式调用是垂直执行的,map这里只需要执行两次。相对于水平执行来说,map会执行尽可能少的次数,而不是把所有元素都 map 转换一遍。

四、中间操作顺序这么重要?

下面的例子由两个中间操作mapfilter,以及一个终端操作forEach组成。让我们再来看看这些操作是如何执行的:

  1. Stream.of("d2", "a2", "b1", "b3", "c")
  2. .map(s -> {
  3. System.out.println("map: " + s);
  4. return s.toUpperCase(); // 转大写
  5. })
  6. .filter(s -> {
  7. System.out.println("filter: " + s);
  8. return s.startsWith("A"); // 过滤出以 A 为前缀的元素
  9. })
  10. .forEach(s -> System.out.println("forEach: " + s)); // for 循环输出
  11. // map: d2
  12. // filter: D2
  13. // map: a2
  14. // filter: A2
  15. // forEach: A2
  16. // map: b1
  17. // filter: B1
  18. // map: b3
  19. // filter: B3
  20. // map: c
  21. // filter: C

学习了上面一小节,您应该已经知道了,mapfilter会对集合中的每个字符串调用五次,如果我们要筛选出A2的话,forEach却只会调用一次,因为只有 “a2” 满足过滤条件。

如果我们改变中间操作的顺序,将filter移动到链头的最开始,就可以大大减少实际的执行次数:

  1. Stream.of("d2", "a2", "b1", "b3", "c")
  2. .filter(s -> {
  3. System.out.println("filter: " + s)
  4. return s.startsWith("a"); // 过滤出以 a 为前缀的元素
  5. })
  6. .map(s -> {
  7. System.out.println("map: " + s);
  8. return s.toUpperCase(); // 转大写
  9. })
  10. .forEach(s -> System.out.println("forEach: " + s)); // for 循环输出
  11. // filter: d2
  12. // filter: a2
  13. // map: a2
  14. // forEach: A2
  15. // filter: b1
  16. // filter: b3
  17. // filter: c

现在,map仅仅只需调用一次,性能得到了提升,这种小技巧对于流中存在大量元素来说,是非常很有用的。

接下来,让我们对上面的代码再添加一个中间操作sorted

  1. Stream.of("d2", "a2", "b1", "b3", "c")
  2. .sorted((s1, s2) -> {
  3. System.out.printf("sort: %s; %s\n", s1, s2);
  4. return s1.compareTo(s2); // 排序
  5. })
  6. .filter(s -> {
  7. System.out.println("filter: " + s);
  8. return s.startsWith("a"); // 过滤出以 a 为前缀的元素
  9. })
  10. .map(s -> {
  11. System.out.println("map: " + s);
  12. return s.toUpperCase(); // 转大写
  13. })
  14. .forEach(s -> System.out.println("forEach: " + s)); // for 循环输出

sorted 是一个有状态的操作,因为它需要在处理的过程中,保存状态以对集合中的元素进行排序。

执行上面代码,输出如下:

  1. sort: a2; d2
  2. sort: b1; a2
  3. sort: b1; d2
  4. sort: b1; a2
  5. sort: b3; b1
  6. sort: b3; d2
  7. sort: c; b3
  8. sort: c; d2
  9. filter: a2
  10. map: a2
  11. forEach: A2
  12. filter: b1
  13. filter: b3
  14. filter: c
  15. filter: d2

咦咦咦?这次怎么又不是垂直执行了。你需要知道的是,sorted是水平执行的。因此,在这种情况下,sorted会对集合中的元素组合调用八次。这里,我们也可以利用上面说道的优化技巧,将 filter 过滤中间操作移动到开头部分:

  1. Stream.of("d2", "a2", "b1", "b3", "c")
  2. .filter(s -> {
  3. System.out.println("filter: " + s);
  4. return s.startsWith("a");
  5. })
  6. .sorted((s1, s2) -> {
  7. System.out.printf("sort: %s; %s\n", s1, s2);
  8. return s1.compareTo(s2);
  9. })
  10. .map(s -> {
  11. System.out.println("map: " + s);
  12. return s.toUpperCase();
  13. })
  14. .forEach(s -> System.out.println("forEach: " + s));
  15. // filter: d2
  16. // filter: a2
  17. // filter: b1
  18. // filter: b3
  19. // filter: c
  20. // map: a2
  21. // forEach: A2

从上面的输出中,我们看到了 sorted从未被调用过,因为经过filter过后的元素已经减少到只有一个,这种情况下,是不用执行排序操作的。因此性能被大大提高了。

六、高级操作

Streams 支持的操作很丰富,除了上面介绍的这些比较常用的中间操作,如filtermap(参见Stream Javadoc)外。还有一些更复杂的操作,如collectflatMap以及reduce。接下来,就让我们学习一下:

本小节中的大多数代码示例均会使用以下 List<Person>进行演示:

  1. class Person {
  2. String name;
  3. int age;
  4. Person(String name, int age) {
  5. this.name = name;
  6. this.age = age;
  7. }
  8. @Override
  9. public String toString() {
  10. return name;
  11. }
  12. }
  13. // 构建一个 Person 集合
  14. List<Person> persons =
  15. Arrays.asList(
  16. new Person("Max", 18),
  17. new Person("Peter", 23),
  18. new Person("Pamela", 23),
  19. new Person("David", 12));

6.1 Collect

collect 是一个非常有用的终端操作,它可以将流中的元素转变成另外一个不同的对象,例如一个ListSetMap。collect 接受入参为Collector(收集器),它由四个不同的操作组成:供应器(supplier)、累加器(accumulator)、组合器(combiner)和终止器(finisher)。
这些都是个啥?别慌,看上去非常复杂的样子,但好在大多数情况下,您并不需要自己去实现收集器。因为 Java 8通过Collectors类内置了各种常用的收集器,你直接拿来用就行了。
让我们先从一个非常常见的用例开始:

  1. List<Person> filtered =
  2. persons
  3. .stream() // 构建流
  4. .filter(p -> p.name.startsWith("P")) // 过滤出名字以 P 开头的
  5. .collect(Collectors.toList()); // 生成一个新的 List
  6. System.out.println(filtered); // [Peter, Pamela]

你也看到了,从流中构造一个 List 异常简单。如果说你需要构造一个 Set 集合,只需要使用Collectors.toSet()就可以了。

接下来这个示例,将会按年龄对所有人进行分组:

  1. Map<Integer, List<Person>> personsByAge = persons
  2. .stream()
  3. .collect(Collectors.groupingBy(p -> p.age)); // 以年龄为 key,进行分组
  4. personsByAge
  5. .forEach((age, p) -> System.out.format("age %s: %s\n", age, p));
  6. // age 18: [Max]
  7. // age 23: [Peter, Pamela]
  8. // age 12: [David]

除了上面这些操作。您还可以在流上执行聚合操作,例如,计算所有人的平均年龄:

  1. Double averageAge = persons
  2. .stream()
  3. .collect(Collectors.averagingInt(p -> p.age)); // 聚合出平均年龄
  4. System.out.println(averageAge); // 19.0

如果您还想得到一个更全面的统计信息,摘要收集器可以返回一个特殊的内置统计对象。通过它,我们可以简单地计算出最小年龄、最大年龄、平均年龄、总和以及总数量。

  1. IntSummaryStatistics ageSummary =
  2. persons
  3. .stream()
  4. .collect(Collectors.summarizingInt(p -> p.age)); // 生成摘要统计
  5. System.out.println(ageSummary);
  6. // IntSummaryStatistics{count=4, sum=76, min=12, average=19.000000, max=23}

下一个这个示例,可以将所有人名连接成一个字符串:

  1. String phrase = persons
  2. .stream()
  3. .filter(p -> p.age >= 18) // 过滤出年龄大于等于18的
  4. .map(p -> p.name) // 提取名字
  5. .collect(Collectors.joining(" and ", "In Germany ", " are of legal age.")); // 以 In Germany 开头,and 连接各元素,再以 are of legal age. 结束
  6. System.out.println(phrase);
  7. // In Germany Max and Peter and Pamela are of legal age.

连接收集器的入参接受分隔符,以及可选的前缀以及后缀。

对于如何将流转换为 Map集合,我们必须指定 Map 的键和值。这里需要注意,Map 的键必须是唯一的,否则会抛出IllegalStateException 异常。

你可以选择传递一个合并函数作为额外的参数来避免发生这个异常:

  1. Map<Integer, String> map = persons
  2. .stream()
  3. .collect(Collectors.toMap(
  4. p -> p.age,
  5. p -> p.name,
  6. (name1, name2) -> name1 + ";" + name2)); // 对于同样 key 的,将值拼接
  7. System.out.println(map);
  8. // {18=Max, 23=Peter;Pamela, 12=David}

既然我们已经知道了这些强大的内置收集器,接下来就让我们尝试构建自定义收集器吧。

比如说,我们希望将流中的所有人转换成一个字符串,包含所有大写的名称,并以|分割。为了达到这种效果,我们需要通过Collector.of()创建一个新的收集器。同时,我们还需要传入收集器的四个组成部分:供应器、累加器、组合器和终止器。

  1. Collector<Person, StringJoiner, String> personNameCollector =
  2. Collector.of(
  3. () -> new StringJoiner(" | "), // supplier 供应器
  4. (j, p) -> j.add(p.name.toUpperCase()), // accumulator 累加器
  5. (j1, j2) -> j1.merge(j2), // combiner 组合器
  6. StringJoiner::toString); // finisher 终止器
  7. String names = persons
  8. .stream()
  9. .collect(personNameCollector); // 传入自定义的收集器
  10. System.out.println(names); // MAX | PETER | PAMELA | DAVID

由于Java 中的字符串是 final 类型的,我们需要借助辅助类StringJoiner,来帮我们构造字符串。

最开始供应器使用分隔符构造了一个StringJointer

累加器用于将每个人的人名转大写,然后加到StringJointer中。

组合器将两个StringJointer合并为一个。

最终,终结器从StringJointer构造出预期的字符串。

6.2 FlatMap

上面我们已经学会了如通过map操作, 将流中的对象转换为另一种类型。但是,Map只能将每个对象映射到另一个对象。

如果说,我们想要将一个对象转换为多个其他对象或者根本不做转换操作呢?这个时候,flatMap就派上用场了。

FlatMap 能够将流的每个元素, 转换为其他对象的流。因此,每个对象可以被转换为零个,一个或多个其他对象,并以流的方式返回。之后,这些流的内容会被放入flatMap返回的流中。

在学习如何实际操作flatMap之前,我们先新建两个类,用来测试:

  1. class Foo {
  2. String name;
  3. List<Bar> bars = new ArrayList<>();
  4. Foo(String name) {
  5. this.name = name;
  6. }
  7. }
  8. class Bar {
  9. String name;
  10. Bar(String name) {
  11. this.name = name;
  12. }
  13. }
  14. 复制代码

接下来,通过我们上面学习到的流知识,来实例化一些对象:

  1. List<Foo> foos = new ArrayList<>();
  2. // 创建 foos 集合
  3. IntStream
  4. .range(1, 4)
  5. .forEach(i -> foos.add(new Foo("Foo" + i)));
  6. // 创建 bars 集合
  7. foos.forEach(f ->
  8. IntStream
  9. .range(1, 4)
  10. .forEach(i -> f.bars.add(new Bar("Bar" + i + " <- " + f.name))));
  11. 复制代码

我们创建了包含三个foo的集合,每个foo中又包含三个 bar

flatMap 的入参接受一个返回对象流的函数。为了处理每个foo中的bar,我们需要传入相应 stream 流:

  1. foos.stream()
  2. .flatMap(f -> f.bars.stream())
  3. .forEach(b -> System.out.println(b.name));
  4. // Bar1 <- Foo1
  5. // Bar2 <- Foo1
  6. // Bar3 <- Foo1
  7. // Bar1 <- Foo2
  8. // Bar2 <- Foo2
  9. // Bar3 <- Foo2
  10. // Bar1 <- Foo3
  11. // Bar2 <- Foo3
  12. // Bar3 <- Foo3
  13. 复制代码

如上所示,我们已成功将三个 foo对象的流转换为九个bar对象的流。

最后,上面的这段代码可以简化为单一的流式操作:

  1. IntStream.range(1, 4)
  2. .mapToObj(i -> new Foo("Foo" + i))
  3. .peek(f -> IntStream.range(1, 4)
  4. .mapToObj(i -> new Bar("Bar" + i + " <- " f.name))
  5. .forEach(f.bars::add))
  6. .flatMap(f -> f.bars.stream())
  7. .forEach(b -> System.out.println(b.name));
  8. 复制代码

flatMap也可用于Java8引入的Optional类。OptionalflatMap操作返回一个Optional或其他类型的对象。所以它可以用于避免繁琐的null检查。

接下来,让我们创建层次更深的对象:

  1. class Outer {
  2. Nested nested;
  3. }
  4. class Nested {
  5. Inner inner;
  6. }
  7. class Inner {
  8. String foo;
  9. }
  10. 复制代码

为了处理从 Outer 对象中获取最底层的 foo 字符串,你需要添加多个null检查来避免可能发生的NullPointerException,如下所示:

  1. Outer outer = new Outer();
  2. if (outer != null && outer.nested != null && outer.nested.inner != null) {
  3. System.out.println(outer.nested.inner.foo);
  4. }
  5. 复制代码

我们还可以使用OptionalflatMap操作,来完成上述相同功能的判断,且更加优雅:

  1. Optional.of(new Outer())
  2. .flatMap(o -> Optional.ofNullable(o.nested))
  3. .flatMap(n -> Optional.ofNullable(n.inner))
  4. .flatMap(i -> Optional.ofNullable(i.foo))
  5. .ifPresent(System.out::println);
  6. 复制代码

如果不为空的话,每个flatMap的调用都会返回预期对象的Optional包装,否则返回为nullOptional包装类。

6.3Reduce

规约操作可以将流的所有元素组合成一个结果。Java 8 支持三种不同的reduce方法。第一种将流中的元素规约成流中的一个元素。
让我们看看如何使用这种方法,来筛选出年龄最大的那个人:

  1. persons
  2. .stream()
  3. .reduce((p1, p2) -> p1.age > p2.age ? p1 : p2)
  4. .ifPresent(System.out::println); // Pamela
  5. 复制代码

reduce方法接受BinaryOperator积累函数。该函数实际上是两个操作数类型相同的BiFunctionBiFunction功能和Function一样,但是它接受两个参数。示例代码中,我们比较两个人的年龄,来返回年龄较大的人。
第二种reduce方法接受标识值和BinaryOperator累加器。此方法可用于构造一个新的 Person,其中包含来自流中所有其他人的聚合名称和年龄:

  1. Person result =
  2. persons
  3. .stream()
  4. .reduce(new Person("", 0), (p1, p2) -> {
  5. p1.age += p2.age;
  6. p1.name += p2.name;
  7. return p1;
  8. });
  9. System.out.format("name=%s; age=%s", result.name, result.age);
  10. // name=MaxPeterPamelaDavid; age=76
  11. 复制代码

第三种reduce方法接受三个参数:标识值,BiFunction累加器和类型的组合器函数BinaryOperator。由于初始值的类型不一定为Person,我们可以使用这个归约函数来计算所有人的年龄总和:

  1. Integer ageSum = persons
  2. .stream()
  3. .reduce(0, (sum, p) -> sum += p.age, (sum1, sum2) -> sum1 + sum2);
  4. System.out.println(ageSum); // 76
  5. 复制代码

结果为76,但是内部究竟发生了什么呢?让我们再打印一些调试日志:

  1. Integer ageSum = persons
  2. .stream()
  3. .reduce(0,
  4. (sum, p) -> {
  5. System.out.format("accumulator: sum=%s; person=%s\n", sum, p);
  6. return sum += p.age;
  7. },
  8. (sum1, sum2) -> {
  9. System.out.format("combiner: sum1=%s; sum2=%s\n", sum1, sum2);
  10. return sum1 + sum2;
  11. });
  12. // accumulator: sum=0; person=Max
  13. // accumulator: sum=18; person=Peter
  14. // accumulator: sum=41; person=Pamela
  15. // accumulator: sum=64; person=David
  16. 复制代码

你可以看到,累加器函数完成了所有工作。它首先使用初始值0和第一个人年龄相加。接下来的三步中sum会持续增加,直到76。
等等?好像哪里不太对!组合器从来都没有调用过啊?
我们以并行流的方式运行上面的代码,看看日志输出:

  1. Integer ageSum = persons
  2. .parallelStream()
  3. .reduce(0,
  4. (sum, p) -> {
  5. System.out.format("accumulator: sum=%s; person=%s\n", sum, p);
  6. return sum += p.age;
  7. },
  8. (sum1, sum2) -> {
  9. System.out.format("combiner: sum1=%s; sum2=%s\n", sum1, sum2);
  10. return sum1 + sum2;
  11. });
  12. // accumulator: sum=0; person=Pamela
  13. // accumulator: sum=0; person=David
  14. // accumulator: sum=0; person=Max
  15. // accumulator: sum=0; person=Peter
  16. // combiner: sum1=18; sum2=23
  17. // combiner: sum1=23; sum2=12
  18. // combiner: sum1=41; sum2=35
  19. 复制代码

并行流的执行方式完全不同。这里组合器被调用了。实际上,由于累加器被并行调用,组合器需要被用于计算部分累加值的总和。
让我们在下一章深入探讨并行流。

七、并行流

流是可以并行执行的,当流中存在大量元素时,可以显著提升性能。并行流底层使用的ForkJoinPool, 它由ForkJoinPool.commonPool()方法提供。底层线程池的大小最多为五个 - 具体取决于 CPU 可用核心数:

  1. ForkJoinPool commonPool = ForkJoinPool.commonPool();
  2. System.out.println(commonPool.getParallelism()); // 3
  3. 复制代码

在我的机器上,公共池初始化默认值为 3。你也可以通过设置以下JVM参数可以减小或增加此值:

  1. -Djava.util.concurrent.ForkJoinPool.common.parallelism=5
  2. 复制代码

集合支持parallelStream()方法来创建元素的并行流。或者你可以在已存在的数据流上调用中间方法parallel(),将串行流转换为并行流,这也是可以的。
为了详细了解并行流的执行行为,我们在下面的示例代码中,打印当前线程的信息:

  1. Arrays.asList("a1", "a2", "b1", "c2", "c1")
  2. .parallelStream()
  3. .filter(s -> {
  4. System.out.format("filter: %s [%s]\n",
  5. s, Thread.currentThread().getName());
  6. return true;
  7. })
  8. .map(s -> {
  9. System.out.format("map: %s [%s]\n",
  10. s, Thread.currentThread().getName());
  11. return s.toUpperCase();
  12. })
  13. .forEach(s -> System.out.format("forEach: %s [%s]\n",
  14. s, Thread.currentThread().getName()));
  15. 复制代码

通过日志输出,我们可以对哪个线程被用于执行流式操作,有个更深入的理解:

  1. filter: b1 [main]
  2. filter: a2 [ForkJoinPool.commonPool-worker-1]
  3. map: a2 [ForkJoinPool.commonPool-worker-1]
  4. filter: c2 [ForkJoinPool.commonPool-worker-3]
  5. map: c2 [ForkJoinPool.commonPool-worker-3]
  6. filter: c1 [ForkJoinPool.commonPool-worker-2]
  7. map: c1 [ForkJoinPool.commonPool-worker-2]
  8. forEach: C2 [ForkJoinPool.commonPool-worker-3]
  9. forEach: A2 [ForkJoinPool.commonPool-worker-1]
  10. map: b1 [main]
  11. forEach: B1 [main]
  12. filter: a1 [ForkJoinPool.commonPool-worker-3]
  13. map: a1 [ForkJoinPool.commonPool-worker-3]
  14. forEach: A1 [ForkJoinPool.commonPool-worker-3]
  15. forEach: C1 [ForkJoinPool.commonPool-worker-2]
  16. 复制代码

如您所见,并行流使用了所有的ForkJoinPool中的可用线程来执行流式操作。在持续的运行中,输出结果可能有所不同,因为所使用的特定线程是非特定的。
让我们通过添加中间操作sort来扩展上面示例:

  1. Arrays.asList("a1", "a2", "b1", "c2", "c1")
  2. .parallelStream()
  3. .filter(s -> {
  4. System.out.format("filter: %s [%s]\n",
  5. s, Thread.currentThread().getName());
  6. return true;
  7. })
  8. .map(s -> {
  9. System.out.format("map: %s [%s]\n",
  10. s, Thread.currentThread().getName());
  11. return s.toUpperCase();
  12. })
  13. .sorted((s1, s2) -> {
  14. System.out.format("sort: %s <> %s [%s]\n",
  15. s1, s2, Thread.currentThread().getName());
  16. return s1.compareTo(s2);
  17. })
  18. .forEach(s -> System.out.format("forEach: %s [%s]\n",
  19. s, Thread.currentThread().getName()));
  20. 复制代码

运行代码,输出结果看上去有些奇怪:

  1. filter: c2 [ForkJoinPool.commonPool-worker-3]
  2. filter: c1 [ForkJoinPool.commonPool-worker-2]
  3. map: c1 [ForkJoinPool.commonPool-worker-2]
  4. filter: a2 [ForkJoinPool.commonPool-worker-1]
  5. map: a2 [ForkJoinPool.commonPool-worker-1]
  6. filter: b1 [main]
  7. map: b1 [main]
  8. filter: a1 [ForkJoinPool.commonPool-worker-2]
  9. map: a1 [ForkJoinPool.commonPool-worker-2]
  10. map: c2 [ForkJoinPool.commonPool-worker-3]
  11. sort: A2 <> A1 [main]
  12. sort: B1 <> A2 [main]
  13. sort: C2 <> B1 [main]
  14. sort: C1 <> C2 [main]
  15. sort: C1 <> B1 [main]
  16. sort: C1 <> C2 [main]
  17. forEach: A1 [ForkJoinPool.commonPool-worker-1]
  18. forEach: C2 [ForkJoinPool.commonPool-worker-3]
  19. forEach: B1 [main]
  20. forEach: A2 [ForkJoinPool.commonPool-worker-2]
  21. forEach: C1 [ForkJoinPool.commonPool-worker-1]
  22. 复制代码

貌似sort只在主线程上串行执行。但是实际上,并行流中的sort在底层使用了Java8中新的方法Arrays.parallelSort()。如 javadoc官方文档解释的,这个方法会按照数据长度来决定以串行方式,或者以并行的方式来执行。

如果指定数据的长度小于最小数值,它则使用相应的Arrays.sort方法来进行排序。

回到上小节 reduce的例子。我们已经发现了组合器函数只在并行流中调用,而不不会在串行流中被调用。
让我们来实际观察一下涉及到哪个线程:

  1. List<Person> persons = Arrays.asList(
  2. new Person("Max", 18),
  3. new Person("Peter", 23),
  4. new Person("Pamela", 23),
  5. new Person("David", 12));
  6. persons
  7. .parallelStream()
  8. .reduce(0,
  9. (sum, p) -> {
  10. System.out.format("accumulator: sum=%s; person=%s [%s]\n",
  11. sum, p, Thread.currentThread().getName());
  12. return sum += p.age;
  13. },
  14. (sum1, sum2) -> {
  15. System.out.format("combiner: sum1=%s; sum2=%s [%s]\n",
  16. sum1, sum2, Thread.currentThread().getName());
  17. return sum1 + sum2;
  18. });
  19. 复制代码

通过控制台日志输出,累加器和组合器均在所有可用的线程上并行执行:

  1. accumulator: sum=0; person=Pamela; [main]
  2. accumulator: sum=0; person=Max; [ForkJoinPool.commonPool-worker-3]
  3. accumulator: sum=0; person=David; [ForkJoinPool.commonPool-worker-2]
  4. accumulator: sum=0; person=Peter; [ForkJoinPool.commonPool-worker-1]
  5. combiner: sum1=18; sum2=23; [ForkJoinPool.commonPool-worker-1]
  6. combiner: sum1=23; sum2=12; [ForkJoinPool.commonPool-worker-2]
  7. combiner: sum1=41; sum2=35; [ForkJoinPool.commonPool-worker-2]
  8. 复制代码

总之,你需要记住的是,并行流对含有大量元素的数据流提升性能极大。但是你也需要记住并行流的一些操作,例如reducecollect操作,需要额外的计算(如组合操作),这在串行执行时是并不需要。
此外,我们也了解了,所有并行流操作都共享相同的 JVM 相关的公共ForkJoinPool。所以你可能需要避免写出一些又慢又卡的流式操作,这很有可能会拖慢你应用中,严重依赖并行流的其它部分代码的性能。