Java8新特性Stream流详解

迷南。 2023-09-24 13:53 216阅读 0赞
  • Stream流是Java8新特性中实践起来最让人舒服的功能，它让我们告别了繁琐的for循环迭代，所以Stream是我们必须要掌握的一个技术特性
  • 此文将带着你如何使用 Java 8 不同类型的 Stream 操作。同时您还将了解流的处理顺序，以及不同顺序的流操作是如何影响运行时性能的。

① Stream 流是如何工作的

流表示包含着一系列元素的集合，我们可以对其做不同类型的操作，用来对这些元素执行计算。

代码示例：

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

我们可以对流进行中间操作或者终端操作。

  • 中间操作会再次返回一个流，所以，我们可以链接多个中间操作，注意这里是不用加分号的。上图中的filter 过滤，map 对象转换，sorted 排序，就属于中间操作。

  • 终端操作是对流操作的一个结束动作，一般返回 void 或者一个非流的结果。上图中的 forEach循环 就是一个终止操作。

    [外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-qIF7HLZW-1667372662338)(Java8新特性Stream流详解.assets/155602207084647.jpeg)]

实际上，大部分流操作都支持 lambda 表达式作为参数，正确理解，应该说是接受一个函数式接口的实现作为参数。

② 不同类型的Stream流

我们可以通过从各种数据源中创建 Stream 流，其中以 Collection 集合最为常见。如 List 和 Set 均支持 stream() 方法来创建顺序流或者是并行流。并行流是通过多线程的方式来执行的，它能够充分发挥多核 CPU 的优势来提升性能。

在集合上调用stream()方法会返回一个普通的 Stream 流。但是我们也可以通过 Stream.of() 从一堆对象中创建 Stream 流。

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

不过除了常规对象流之外，Java 8还附带了一些特殊类型的流，用于处理原始数据类型int，long以及double。所对应的流是IntStream，LongStream还有DoubleStream。

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

  • 原始类型流使用其独有的函数式接口，例如IntFunction代替Function，IntPredicate代替Predicate。

  • 原始类型流支持额外的终端聚合操作，sum()以及average()，如下所示：

    1. Arrays.stream(new int[] {
    2.       1, 2, 3})
    3.     .map(n -> 2 * n + 1) // 对数值中的每个对象执行 2*n + 1 操作
    4.     .average() // 求平均值
    5.     .ifPresent(System.out::println);  // 如果值不为空，则输出
    6. // 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 循环打印
  5. // a1
  6. // a2
  7. // a3

代码示例

  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 循环打印
  6. // a1
  7. // a2
  8. // a3

③ Stream流的处理顺序

首选声明，中间操作是有延迟性的。

代码示例：

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

上述代码带有欺骗性你可能会认为，将依次打印 “d2”, “a2”, “b1”, “b3”, “c” 元素。其实不会输出任何东西。

因为当且仅当存在终端操作时，中间操作操作才会被执行。

  1. Stream.of("d2", "a2", "b1", "b3", "c")
  2.     .filter(s -> {
  4.         System.out.println("filter: " + s);
  5.         return true;
  6.     })
  7.     .forEach(s -> System.out.println("forEach: " + s));
  9. // 输出
  10. filter: d2
  11. forEach: d2
  12. filter: a2
  13. forEach: a2
  14. filter: b1
  15. forEach: b1
  16. filter: b3
  17. forEach: b3
  18. filter: c
  19. forEach: c

这样就能输出所有元素。

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

原因是出于性能的考虑。这样设计可以减少对每个元素的实际操作数。

代码示例：

  1. @Test
  2. void contextLoads() {
  4.     Stream.of("d2", "a2", "b1", "b3", "c")
  5.         .map(s -> {
  7.             System.out.println("map: " + s);
  8.             return s.toUpperCase(); // 转大写
  9.         })
  10.         .anyMatch(s -> {
  12.             System.out.println("anyMatch: " + s);
  13.             return s.startsWith("A"); // 过滤出以 A 为前缀的元素
  14.         });
  15.     // 输出结果
  16.     // map:      d2
  17.     // anyMatch: D2
  18.     // map:      a2
  19.     // anyMatch: A2
  20. }

从此可以明显的看出输出的结果是随着链条垂直移动的。

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

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

④ 中间操作属性呢重要吗？

代码示例

例子由两个中间操作map和filter，以及一个终端操作forEach组成

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

我们知道了中间操作的执行顺序，那么我们就知道map和filter会对集合中的每个字符串调用五次，而forEach却只会调用一次，因为只有 “a2” 满足过滤条件。

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

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

现在，map仅仅只需调用一次，性能得到了提升。

如果在对上述例子增加sorted操作，那么对性能影响更大。

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

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

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

通过结果能发现，sorted是水平执行的。因此，在这种情况下，sorted会对集合中的元素组合调用八次。

我们把sorted操作放在过滤后面，那么性能就能大大提升

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

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

⑤ 数据流复用问题

Java8 Stream 流是不能被复用的，一旦你调用任何终端操作，流就会关闭。

  1. Stream<String> stream =
  2.     Stream.of("d2", "a2", "b1", "b3", "c")
  3.         .filter(s -> s.startsWith("a"));
  5. stream.anyMatch(s -> true);    // ok
  6. stream.noneMatch(s -> true);   // 报异常exception

当我们对流进行了终止操作后，流就会关闭，之后不能去操作流了。再操作会报异常。

  1. java.lang.IllegalStateException: stream has already been operated upon or closed
  2.     at java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:229)
  3.     at java.util.stream.ReferencePipeline.anyMatch(ReferencePipeline.java:449)
  4.     at cn.cvzhanshi.test.TestApplicationTests.contextLoads(TestApplicationTests.java:33)
  5.     at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
  6.     at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
  7.     at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
  8.     at java.lang.reflect.Method.invoke(Method.java:483)
  9.     at org.junit.platform.commons.util.ReflectionUtils.invokeMethod(ReflectionUtils.java:688)
  10.     at org.junit.jupiter.engine.execution.MethodInvocation.proceed(MethodInvocation.java:60)

但是我们可以通过 Supplier 来包装一下流，通过 get() 方法来构建一个新的 Stream 流。就很好的克制了这个问题。

  1. Supplier<Stream<String>> streamSupplier =
  2.     () -> Stream.of("d2", "a2", "b1", "b3", "c")
  3.             .filter(s -> s.startsWith("a"));
  5. streamSupplier.get().anyMatch(s -> true);   // ok
  6. streamSupplier.get().noneMatch(s -> true);  // ok

⑥ 高级操作

Streams 支持的操作很丰富，除了上面介绍的这些比较常用的中间操作，如filter或map。还有一些更复杂的操作，如collect，flatMap以及reduce。

接下来测试代码均用Person类来测试

  1. /**
  2.  * @author cVzhanshi
  3.  * @create 2022-10-08 10:02
  4.  */
  5. @Data
  6. @AllArgsConstructor
  7. @NoArgsConstructor
  8. public class Person {
  10.     String name;
  11.     Integer age;
  12. }

6.1 Collect

collect 是一个非常有用的终端操作，它可以将流中的元素转变成另外一个不同的对象，例如一个List，Set或Map。collect 接受入参为Collector（收集器），它由四个不同的操作组成：供应器（supplier）、累加器（accumulator）、组合器（combiner）和终止器（finisher）。

示例一：collect(Collectors.toList())

  • 过滤出名字以 P 开头的，然后生成一个新的List

    @Test
    void testCollect(){

    1. // 构建一个 Person 集合
    2. List<Person> persons =
    3.     Arrays.asList(
    4.     new Person("Max", 18),
    5.     new Person("Peter", 23),
    6.     new Person("Pamela", 23),
    7.     new Person("David", 12));
  1.     List<Person> persons2 = persons
  2.         .stream() // 构建流
  3.         .filter(person -> person.getName().startsWith("P")) // 过滤
  4.         .collect(Collectors.toList()); // 收集流的结果，生成一个新的List
  5.     persons2.forEach(System.out::println);
  6. }

示例二：collect(Collectors.groupingBy(p -> p.getName()))

  • 将会按年龄对所有人进行分组

    @Test
    void testCollect(){

    1. // 构建一个 Person 集合
    2. List<Person> persons =
    3.     Arrays.asList(
    4.     new Person("Max", 18),
    5.     new Person("Peter", 23),
    6.     new Person("Pamela", 23),
    7.     new Person("David", 12));
  1.     Map<String, List<Person>> collect = persons
  2.         .stream() // 构建流
  3.         .collect(Collectors.groupingBy(p -> p.getName()));
  4.     collect.forEach((k,v) -> System.out.format("age %s : %s\n" ,k ,v));
  5. }

示例三：collect(Collectors.averagingInt(p -> p.getAge()))

  • 计算所有人的平均年龄

    @Test
    void testCollect(){

    1. // 构建一个 Person 集合
    2. List<Person> persons =
    3.     Arrays.asList(
    4.     new Person("Max", 18),
    5.     new Person("Peter", 23),
    6.     new Person("Pamela", 23),
    7.     new Person("David", 12));
    8. System.out.println(persons.stream().collect(Collectors.averagingInt(p -> p.getAge())));

    }

示例四：collect(Collectors.summarizingInt(p -> p.getAge()))

  • 摘要收集器可以返回一个特殊的内置统计对象。通过它，我们可以简单地计算出最小年龄、最大年龄、平均年龄、总和以及总数量。

    @Test
    void testCollect(){

    1. // 构建一个 Person 集合
    2. List<Person> persons =
    3.     Arrays.asList(
    4.     new Person("Max", 18),
    5.     new Person("Peter", 23),
    6.     new Person("Pamela", 23),
    7.     new Person("David", 12));
    8. IntSummaryStatistics ageSummary =
    9.     persons
    10.     .stream()
    11.     .collect(Collectors.summarizingInt(p -> p.getAge())); // 生成摘要统计
    12. System.out.println(ageSummary);

    }

    // IntSummaryStatistics{count=4, sum=76, min=12, average=19.000000, max=23}

示例五：collect(Collectors.joining(“ 连接 “, “开头”, “结尾”))

  • 将所有人名连接成一个字符串

    @Test
    void testCollect(){

    1. // 构建一个 Person 集合
    2. List<Person> persons =
    3.     Arrays.asList(
    4.     new Person("Max", 18),
    5.     new Person("Peter", 23),
    6.     new Person("Pamela", 23),
    7.     new Person("David", 12));
    8. String phrase = persons
    9.     .stream()
    10.     .filter(p -> p.getAge() >= 18) // 过滤出年龄大于等于18的
    11.     .map(p -> p.getName()) // 提取名字
    12.     .collect(Collectors.joining(" 连接 ", "开头", "结尾"));
    13. System.out.println(phrase); // 开头Max 连接 Peter 连接 Pamela结尾

    }

示例六：对于如何将流转换为 Map集合，我们必须指定 Map 的键和值。

这里需要注意，Map 的键必须是唯一的，否则会抛出IllegalStateException 异常。但是我们可以选择传递一个合并函数作为额外的参数来避免发生这个异常：

  1. @Test
  2. void testCollect(){
  4.     // 构建一个 Person 集合
  5.     List<Person> persons =
  6.         Arrays.asList(
  7.         new Person("Max", 18),
  8.         new Person("Peter", 23),
  9.         new Person("Pamela", 23),
  10.         new Person("David", 12));
  11.     Map<Integer, String> map = persons
  12.         .stream()
  13.         .collect(Collectors.toMap(
  14.             p -> p.getAge(),  // key
  15.             p -> p.getName(), // value
  16.             (name1, name2) -> name1 + ";" + name2)); // 对于同样 key 的，将值拼接
  18.     System.out.println(map);// {18=Max, 23=Peter;Pamela, 12=David}
  19. }

示例七：自定义收集器

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

  1. @Test
  2. void testCollect(){
  4.     // 构建一个 Person 集合
  5.     List<Person> persons =
  6.         Arrays.asList(
  7.         new Person("Max", 18),
  8.         new Person("Peter", 23),
  9.         new Person("Pamela", 23),
  10.         new Person("David", 12));
  11.     Collector<Person, StringJoiner, String> personNameCollector =
  12.         Collector.of(
  13.         () -> new StringJoiner(" | "),          // supplier 供应器
  14.         (j, p) -> j.add(p.getName().toUpperCase()),  // accumulator 累加器
  15.         (j1, j2) -> j1.merge(j2),               // combiner 组合器
  16.         StringJoiner::toString);                // finisher 终止器
  18.     String names = persons
  19.         .stream()
  20.         .collect(personNameCollector); // 传入自定义的收集器
  22.     System.out.println(names);  // MAX | PETER | PAMELA | DAVID
  23. }

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

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

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

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

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

6.2 Reduce

Java 8 支持三种不同的reduce方法。

第一种将流中的元素规约成流中的一个元素。

  • 筛选出年龄最大的那个人

    @Test

    1. void testCollect(){
    3.     // 构建一个 Person 集合
    4.     List<Person> persons =
    5.             Arrays.asList(
    6.                     new Person("Max", 18),
    7.                     new Person("Peter", 23),
    8.                     new Person("Pamela", 23),
    9.                     new Person("David", 12));
    10.     persons
    11.             .stream()
    12.             .reduce((p1, p2) -> p1.getAge() > p2.getAge() ? p1 : p2)
    13.             .ifPresent(System.out::println);    // Pamela
    14. }

reduce方法接受BinaryOperator积累函数。该函数实际上是两个操作数类型相同的BiFunction。BiFunction功能和Function一样，但是它接受两个参数。示例代码中，我们比较两个人的年龄，来返回年龄较大的人。

第二种reduce方法接受标识值和BinaryOperator累加器。此方法可用于构造一个新的 Person，其中包含来自流中所有其他人的聚合名称和年龄

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

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

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

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

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

累加器函数完成了所有工作。它首先使用初始值0和第一个人年龄相加。接下来的三步中sum会持续增加，直到76。但是我们发现组合器好像没有执行一次

用并行流去测试一下

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

并行流的执行方式完全不同。这里组合器被调用了。实际上，由于累加器被并行调用，组合器需要被用于计算部分累加值的总和。

⑦ 并行流

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

  1. @Test
  2. void testParallelStream(){
  4.     ForkJoinPool commonPool = ForkJoinPool.commonPool();
  5.     System.out.println(commonPool.getParallelism());    // 7
  6. }

集合支持parallelStream()方法来创建元素的并行流。或者你可以在已存在的数据流上调用中间方法parallel()，将串行流转换为并行流

为了详细了解并行流的执行行为，我们在下面的示例代码中，打印当前线程的信息：

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

输出日志

  1. filter: b1 [main]
  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: a1 [ForkJoinPool.commonPool-worker-3]
  7. map: a1 [ForkJoinPool.commonPool-worker-3]
  8. map: b1 [main]
  9. forEach: A1 [ForkJoinPool.commonPool-worker-3]
  10. forEach: A2 [ForkJoinPool.commonPool-worker-1]
  11. forEach: C1 [ForkJoinPool.commonPool-worker-2]
  12. filter: c2 [ForkJoinPool.commonPool-worker-4]
  13. map: c2 [ForkJoinPool.commonPool-worker-4]
  14. forEach: C2 [ForkJoinPool.commonPool-worker-4]
  15. forEach: B1 [main]

并行流使用了所有的ForkJoinPool中的可用线程来执行流式操作。在持续的运行中，输出结果可能有所不同，因为所使用的特定线程是非特定的。

  • 添加中间操作sort来扩展上面示例：

    Arrays.asList(“a1”, “a2”, “b1”, “c2”, “c1”)

    1. .parallelStream()
    2. .filter(s -> {
    4.     System.out.format("filter: %s [%s]\n",
    5.                       s, Thread.currentThread().getName());
    6.     return true;
    7. })
    8. .map(s -> {
    10.     System.out.format("map: %s [%s]\n",
    11.                       s, Thread.currentThread().getName());
    12.     return s.toUpperCase();
    13. })
    14. .sorted((s1, s2) -> {
    16.     System.out.format("sort: %s <> %s [%s]\n",
    17.                       s1, s2, Thread.currentThread().getName());
    18.     return s1.compareTo(s2);
    19. })
    20. .forEach(s -> System.out.format("forEach: %s [%s]\n",
    21.                                 s, Thread.currentThread().getName()));
  1. // 输出
  2. filter:  c2 [ForkJoinPool.commonPool-worker-3]
  3. filter:  c1 [ForkJoinPool.commonPool-worker-2]
  4. map:     c1 [ForkJoinPool.commonPool-worker-2]
  5. filter:  a2 [ForkJoinPool.commonPool-worker-1]
  6. map:     a2 [ForkJoinPool.commonPool-worker-1]
  7. filter:  b1 [main]
  8. map:     b1 [main]
  9. filter:  a1 [ForkJoinPool.commonPool-worker-2]
  10. map:     a1 [ForkJoinPool.commonPool-worker-2]
  11. map:     c2 [ForkJoinPool.commonPool-worker-3]
  12. sort:    A2 <> A1 [main]
  13. sort:    B1 <> A2 [main]
  14. sort:    C2 <> B1 [main]
  15. sort:    C1 <> C2 [main]
  16. sort:    C1 <> B1 [main]
  17. sort:    C1 <> C2 [main]
  18. forEach: A1 [ForkJoinPool.commonPool-worker-1]
  19. forEach: C2 [ForkJoinPool.commonPool-worker-3]
  20. forEach: B1 [main]
  21. forEach: A2 [ForkJoinPool.commonPool-worker-2]
  22. forEach: C1 [ForkJoinPool.commonPool-worker-1]

貌似sort只在主线程上串行执行。但是实际上，并行流中的sort在底层使用了Java8中新的方法Arrays.parallelSort()。这个方法会按照数据长度来决定以串行方式，或者以并行的方式来执行。（如果指定数据的长度小于最小数值，它则使用相应的Arrays.sort方法来进行排序。）

回到上小节 reduce的例子。我们已经发现了组合器函数只在并行流中调用，而不不会在串行流中被调用。

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

参考文章直达

发表评论

表情：
评论列表 （有 0 条评论，216人围观）

还没有评论，来说两句吧...

相关阅读

    相关 Java 8新特性：流（Stream）

    在Java 8中，引入了新的数据处理方式——流（Stream）。流的概念类似于物理学中的流动，能够以一种高效、灵活的方式对大量数据进行操作。 以下是一些流的主要特性： 1.

    向右看齐向右看齐/ 0 赞/ 99 阅读

    相关 Java8新特性 stream流

    JDK1.8的新特性 1. 前言 JDK1.8已经发布很久了，在很多企业中都已经在使用。并且Spring5、SpringBoot2.0都推荐使用JDK1.8以上版本

    朱雀朱雀/ 0 赞/ 188 阅读

    相关 Java8新特性Stream流详解

    > Stream流是Java8新特性中实践起来最让人舒服的功能，它让我们告别了繁琐的for循环迭代，所以Stream是我们必须要掌握的一个技术特性 > 此文将带着你

    迷南。迷南。/ 0 赞/ 217 阅读

    相关 Java8新特性Stream流详解

    > 陈老老老板 说明：新的专栏，本专栏专门讲Java8新特性，把平时遇到的问题与Java8的写法进行总结，需要注意的地方都标红了，一起加油。 > 本文是介绍Java8新特性

    桃扇骨桃扇骨/ 0 赞/ 226 阅读