java8数据流再识

 3,598浏览     wenqyAdmin   

数据流操作要么是衔接操作，要么是终止操作。当一个函数不修改数据流的底层数据源，它就是无干扰的。当一个函数的操作的执行是确定性的，它就是无状态的。

数据流可以从多种数据源创建，尤其是集合。可以有不同类型的数据流。

  1. /**
  2.      * 从多种数据源创建数据流
  3.      * 
  4.      * @author wenqy
  5.      * @date 2020年1月17日 上午11:00:15
  6.      */
  7.     private void diffStreamType() {
  8.         System.out.println(“—–>diffStreamType—–>”);
  9.         Arrays.asList(“a1”, “a2”, “a3”)
  10.             .stream()
  11.             .findFirst()
  12.             .ifPresent(System.out::println);  // a1
  14.         Stream.of(“a1”, “a2”, “a3”)
  15.             .findFirst()
  16.             .ifPresent(System.out::println);  // a1
  18.         IntStream.range(1, 4)   // 基本数据类型 // 1 2 3
  19.             .forEach(System.out::println);
  20.     }

基本数据流和对象数据流间也可以转换

  1. /**
  2.      * 基本数据流操作
  3.      * 
  4.      * @author wenqy
  5.      * @date 2020年1月17日 上午11:01:26
  6.      */
  7.     private void baseStream() {
  8.         System.out.println(“—–>baseStream—–>”);
  9.         Arrays.stream(new int[] {1, 2, 3})
  10.             .map(n -> 2 * n + 1)
  11.             .average() // 终止操作，求平均值
  12.             .ifPresent(System.out::println);  // 5.0
  14.         Stream.of(“a1”, “a2”, “a3”)
  15.             .map(s -> s.substring(1))
  16.             .mapToInt(Integer::parseInt) // 对象数据流转换为基本数据流
  17.             .max()
  18.             .ifPresent(System.out::println);  // 3
  20.         IntStream.range(1, 4)
  21.             .mapToObj(i -> “a” + i) // 基本数据流转换为对象数据流
  22.             .forEach(System.out::println); // a1 a2 a3
  23.     }

数据流处理时，衔接操作的一个重要特性就是延迟性，在调用链上是垂直移动的，减少每个元素上所执行的实际操作数量。

  1. /**
  2.      * 处理顺序
  3.      * 
  4.      * @author wenqy
  5.      * @date 2020年1月17日 上午11:08:33
  6.      */
  7.     private void streamHandleSort() {
  8.         System.out.println(“—–>streamHandleSort—–>”);
  9.         Stream.of(“d2”, “a2”, “b1”, “b3”, “c”)
  10.             .filter(s -> {
  11.                 System.out.println(“filter: “ + s);
  12.                 return true;
  13.             })
  14.             .forEach(s -> System.out.println(“forEach: “ + s));
  15.         // filter:  d2 forEach: d2 … 每个元素在调用链上垂直移动
  17.         Stream.of(“d2”, “a2”, “b1”, “b3”, “c”)
  18.             .map(s -> {
  19.                 System.out.println(“map: “ + s);
  20.                 return s.toUpperCase();
  21.             })
  22.             .anyMatch(s -> {
  23.                 System.out.println(“anyMatch: “ + s);
  24.                 return s.startsWith(“A”);
  25.             });
  26.         // map:d2 anyMatch:D2 map:a2 anyMatch:A2 anyMatch返回true时终止
  28.         Stream.of(“d2”, “a2”, “b1”, “b3”, “c”)
  29.             .map(s -> {
  30.                 System.out.println(“map: “ + s);
  31.                 return s.toUpperCase();
  32.             })
  33.             .filter(s -> {
  34.                 System.out.println(“filter: “ + s);
  35.                 return s.startsWith(“A”);
  36.             })
  37.             .forEach(s -> System.out.println(“forEach: “ + s));
  38.         // map和filter会对底层集合的每个字符串调用五次，而forEach只会调用一次
  40.         Stream.of(“d2”, “a2”, “b1”, “b3”, “c”)
  41.             .filter(s -> {
  42.                 System.out.println(“filter: “ + s);
  43.                 return s.startsWith(“a”);
  44.             })
  45.             .map(s -> {
  46.                 System.out.println(“map: “ + s);
  47.                 return s.toUpperCase();
  48.             })
  49.             .forEach(s -> System.out.println(“forEach: “ + s));
  50.         // filter移动到调用链的顶端  map只会调用一次 执行更快
  52.         Stream.of(“d2”, “a2”, “b1”, “b3”, “c”)
  53.             .sorted((s1, s2) -> {
  54.                 System.out.printf(“sort: %s; %s\n”, s1, s2);
  55.                 return s1.compareTo(s2);
  56.             })
  57.             .filter(s -> {
  58.                 System.out.println(“filter: “ + s);
  59.                 return s.startsWith(“a”);
  60.             })
  61.             .map(s -> {
  62.                 System.out.println(“map: “ + s);
  63.                 return s.toUpperCase();
  64.             })
  65.             .forEach(s -> System.out.println(“forEach: “ + s));
  66.         // 排序是一类特殊的衔接操作。它是有状态的操作，因为你需要在处理中保存状态来对集合中的元素排序
  68.         Stream.of(“d2”, “a2”, “b1”, “b3”, “c”)
  69.             .filter(s -> {
  70.                 System.out.println(“filter: “ + s);
  71.                 return s.startsWith(“a”);
  72.             })
  73.             .sorted((s1, s2) -> {
  74.                 System.out.printf(“sort: %s; %s\n”, s1, s2);
  75.                 return s1.compareTo(s2);
  76.             })
  77.             .map(s -> {
  78.                 System.out.println(“map: “ + s);
  79.                 return s.toUpperCase();
  80.             })
  81.             .forEach(s -> System.out.println(“forEach: “ + s));
  82.         // 重排调用链来优化性能,这个例子中sorted永远不会调用,极大提升性能
  83.     }

java8的数据流不能被复用。一旦你调用了任何终止操作，数据流就关闭了，要克服这个限制，我们需要为每个我们想要执行的终止操作创建新的数据流调用链。例如，我们创建一个数据流供应器，来构建新的数据流，并且设置好所有衔接操作。

  1. /**
  2.      * 复用数据流
  3.      * 
  4.      * @author wenqy
  5.      * @date 2020年1月17日 上午11:40:02
  6.      */
  7.     private void streamReuse() {
  8.         System.out.println(“—–>streamReuse—–>”);
  9.         Stream<String> stream =
  10.                 Stream.of(“d2”, “a2”, “b1”, “b3”, “c”)
  11.                     .filter(s -> s.startsWith(“a”));
  13.         stream.anyMatch(s -> true);    // ok
  14. //      stream.noneMatch(s -> true);   // exception  java.lang.IllegalStateException: stream has already been operated upon or closed
  16.         Supplier<Stream<String>> streamSupplier =
  17.                 () -> Stream.of(“d2”, “a2”, “b1”, “b3”, “c”)
  18.                         .filter(s -> s.startsWith(“a”));
  19.         // 每次对get()的调用都构造了一个新的数据流，我们将其保存来调用终止操作
  20.         streamSupplier.get().anyMatch(s -> true);   // ok
  21.         streamSupplier.get().noneMatch(s -> true);  // ok
  22.     }

collect

collect是非常有用的终止操作，将流中的元素存放在不同类型的结果中，例如List、Set或者Map。collect接受收集器（Collector），它由四个不同的操作组成：供应器（supplier）、累加器（accumulator）、组合器（combiner）和终止器（finisher）。

  1. /**
  2.      * collect是非常有用的终止操作，将流中的元素存放在不同类型的结果中，例如List、Set或者Map。
  3.      * collect接受收集器（Collector），它由四个不同的操作组成：
  4.      *  供应器（supplier）、累加器（accumulator）、组合器（combiner）和终止器（finisher）
  5.      * 
  6.      * @author wenqy
  7.      * @date 2020年1月17日 上午11:45:18
  8.      */
  9.     private void streamCollect() {
  10.         System.out.println(“—–>streamCollect—–>”);
  12.         List<Person> persons = getPersionList();
  14.         List<Person> filtered =
  15.             persons
  16.                 .stream()
  17.                 .filter(p -> p.getFirstName().startsWith(“P”))
  18.                 .collect(Collectors.toList()); // 构造list
  19.         System.out.println(filtered);    // [Person [firstName=Peter, lastName=null, age=23], Person [firstName=Pamela, lastName=null, age=23]]
  21.         Map<Integer, List<Person>> personsByAge = persons
  22.             .stream()
  23.             .collect(Collectors.groupingBy(p -> p.getAge())); // 构造map key: age
  24.         personsByAge
  25.             .forEach((age, p) -> System.out.format(“age %s: %s\n”, age, p));
  27.         IntSummaryStatistics ageSummary =
  28.             persons
  29.                 .stream()
  30.                 .collect(Collectors.summarizingInt(p -> p.getAge()));
  31.         System.out.println(ageSummary); // 统计：简单计算最小年龄、最大年龄、算术平均年龄、总和和数量
  33.         String phrase = persons
  34.             .stream()
  35.             .filter(p -> p.getAge() >= 18)
  36.             .map(p -> p.getFirstName()) // 键必须是唯一的，否则会抛出IllegalStateException异常
  37.             .collect(Collectors.joining(” and “, “In China “, ” are of legal age.”));
  38.         System.out.println(phrase); // 所有人连接为一个字符串
  40.         Collector<Person, StringJoiner, String> personNameCollector =
  41.             Collector.of(
  42.                 () -> new StringJoiner(” | “),          // supplier
  43.                 (j, p) -> j.add(p.getFirstName().toUpperCase()),  // accumulator
  44.                 (j1, j2) -> j1.merge(j2),               // combiner
  45.                 StringJoiner::toString);                // finisher
  46.         String names = persons
  47.             .stream()
  48.             .collect(personNameCollector);
  49.         System.out.println(names);  // MAX | PETER | PAMELA | DAVID
  50.         // 构建自己特殊收集器。将流中的所有人转换为一个字符串，包含所有大写的名称，并以|分割。
  51.     }

flatMap

我们已经了解了如何通过使用map操作，将流中的对象转换为另一种类型。map有时十分受限，因为每个对象只能映射为一个其它对象。但如何我希望将一个对象转换为多个或零个其他对象呢？flatMap这时就会派上用场。

flatMap将流中的每个元素，转换为其它对象的流。所以每个对象会被转换为零个、一个或多个其它对象，以流的形式返回。这些流的内容之后会放进flatMap所返回的流中。

  1. /**
  2.  * flatMap将流中的每个元素，转换为其它对象的流。
  3.  * 所以每个对象会被转换为零个、一个或多个其它对象，以流的形式返回。
  4.  * 这些流的内容之后会放进flatMap所返回的流中
  5.  * 
  6.  * @author wenqy
  7.  * @date 2020年1月17日 下午1:38:09
  8.  */
  9. private void streamFlatMap() {
  10.     System.out.println(“—–>streamFlatMap—–>”);
  11.     List<Foo> foos = new ArrayList<>();
  13.     // create foos
  14.     IntStream
  15.         .range(1, 4)
  16.         .forEach(i -> foos.add(new Foo(“Foo” + i)));
  17.     // create bars
  18.     foos.forEach(f ->
  19.         IntStream
  20.             .range(1, 4)
  21.             .forEach(i -> f.bars.add(new Bar(“Bar” + i + ” <- “ + f.name))));
  22.     foos.stream()
  23.         .flatMap(f -> f.bars.stream())
  24.         .forEach(b -> System.out.println(b.name)); // 将含有三个foo对象中的流转换为含有九个bar对象的流
  26.     IntStream.range(1, 4)
  27.         .mapToObj(i -> new Foo(“Foo” + i))
  28.         .peek(f -> IntStream.range(1, 4)
  29.             .mapToObj(i -> new Bar(“Bar” + i + ” <- “ + f.name))
  30.             .forEach(f.bars::add)) // 简化为流式操作的单一流水线
  31.         .flatMap(f -> f.bars.stream())
  32.         .forEach(b -> System.out.println(b.name));
  34.     Optional.of(new Outer())
  35.         .flatMap(o -> Optional.ofNullable(o.nested))
  36.         .flatMap(n -> Optional.ofNullable(n.inner))
  37.         .flatMap(i -> Optional.ofNullable(i.foo))
  38.         .ifPresent(System.out::println);
  39.     // 如果存在的话，每个flatMap的调用都会返回预期对象的Optional包装，
  40.     // 否则为null的Optional包装,避免潜在NullPointerException
  41. }

reduce

归约操作将所有流中的元素组合为单一结果。Java8支持三种不同类型的reduce方法。

  1. /**
  2.      * 归约操作将所有流中的元素组合为单一结果
  3.      * 
  4.      * @author wenqy
  5.      * @date 2020年1月17日 下午2:01:23
  6.      */
  7.     private void streamReduce() {
  8.         System.out.println(“—–>streamReduce—–>”);
  9.         List<Person> persons = getPersionList();
  10.         persons
  11.             .stream()
  12.             .reduce((p1, p2) -> p1.getAge() > p2.getAge() ? p1 : p2)
  13.             .ifPresent(System.out::println);    // 计算年龄最大的人 Pamela
  15.         Person result =
  16.             persons
  17.                 .stream()
  18.                 .reduce(new Person(“”, 0), (p1, p2) -> {
  19.                     p1.setAge(p1.getAge() + p2.getAge());
  20.                     p1.setFirstName(p1.getFirstName() + p2.getFirstName());
  21.                     return p1;  // 构造带有聚合后名称和年龄的新Person对象
  22.                 });
  23.         // name=MaxPeterPamelaDavid; age=76
  24.         System.out.format(“name=%s; age=%s.\n”, result.getFirstName(), result.getAge());
  26.         Integer ageSum = persons
  27.             .stream()
  28.             .reduce(0, (sum, p) -> sum += p.getAge(), (sum1, sum2) -> sum1 + sum2);
  29.         System.out.println(ageSum);  // 计算所有人的年龄总和 76
  31.         Integer ageSum2 = persons
  32.             .stream()
  33.             .reduce(0,
  34.                 (sum, p) -> {
  35.                     System.out.format(“accumulator: sum=%s; person=%s\n”, sum, p);
  36.                     return sum += p.getAge();
  37.                 },
  38.                 (sum1, sum2) -> {
  39.                     System.out.format(“combiner: sum1=%s; sum2=%s\n”, sum1, sum2);
  40.                     return sum1 + sum2;
  41.                 });
  42.         System.out.println(ageSum2); // 输出调试信息，combiner并没有输出
  44.         Integer ageSum3 = persons
  45.             .parallelStream()
  46.             .reduce(0,
  47.                 (sum, p) -> {
  48.                     System.out.format(“accumulator: sum=%s; person=%s [%s]\n”, sum, p, Thread.currentThread().getName());
  49.                     return sum += p.getAge();
  50.                 },
  51.                 (sum1, sum2) -> {
  52.                     System.out.format(“combiner: sum1=%s; sum2=%s [%s]\n”, sum1, sum2, Thread.currentThread().getName());
  53.                     return sum1 + sum2;
  54.                 });
  55.         System.out.println(ageSum3); // 并行方式
  56.     }

parallelStream

流可以并行执行，在大量输入元素上可以提升运行时的性能。并行流使用公共的ForkJoinPool，由ForkJoinPool.commonPool()方法提供。底层线程池的大小最大为五个线程 — 取决于CPU的物理核数。

组合器函数只在并行流中调用，而不在串行流中调用

  1. /**
  2.      * 并行流
  3.      * 
  4.      * @author wenqy
  5.      * @date 2020年1月17日 下午2:31:57
  6.      */
  7.     private void streamParallel() {
  8.         System.out.println(“—–>streamParallel—–>”);
  9.         ForkJoinPool commonPool = ForkJoinPool.commonPool();
  10.         System.out.println(commonPool.getParallelism());    // 底层线程池的大小 — 取决于CPU的物理核数 本机 默认 7 
  11.         // 可用JVM参数增减 -Djava.util.concurrent.ForkJoinPool.common.parallelism=5
  13.         Arrays.asList(“a1”, “a2”, “b1”, “c2”, “c1”)
  14.             .parallelStream()
  15.             .filter(s -> {
  16.                 System.out.format(“filter: %s [%s]\n”,
  17.                     s, Thread.currentThread().getName());
  18.                 return true;
  19.             })
  20.             .map(s -> {
  21.                 System.out.format(“map: %s [%s]\n”,
  22.                     s, Thread.currentThread().getName());
  23.                 return s.toUpperCase();
  24.             })
  25.             .forEach(s -> System.out.format(“forEach: %s [%s]\n”,
  26.                 s, Thread.currentThread().getName()));
  27.         // 并行流使用了所有公共的ForkJoinPool中的可用线程来执行流式操作
  29.         Arrays.asList(“a1”, “a2”, “b1”, “c2”, “c1”)
  30.             .parallelStream()
  31.             .filter(s -> {
  32.                 System.out.format(“filter: %s [%s]\n”,
  33.                     s, Thread.currentThread().getName());
  34.                 return true;
  35.             })
  36.             .map(s -> {
  37.                 System.out.format(“map: %s [%s]\n”,
  38.                     s, Thread.currentThread().getName());
  39.                 return s.toUpperCase();
  40.             })
  41.             .sorted((s1, s2) -> {
  42.                 System.out.format(“sort: %s <> %s [%s]\n”,
  43.                     s1, s2, Thread.currentThread().getName());
  44.                 return s1.compareTo(s2);
  45.             })
  46.             .forEach(s -> System.out.format(“forEach: %s [%s]\n”,
  47.                 s, Thread.currentThread().getName()));
  48.         // sort看起来只在主线程上串行执行。实际上，并行流上的sort在背后使用了Java8中新的方法Arrays.parallelSort()。
  49.         // 如javadoc所说，这个方法会参照数据长度来决定以串行或并行来执行,如果指定数据的长度小于最小粒度，它使用相应的Arrays.sort方法来排序
  50.         // 所有并行流操作都共享相同的JVM相关的公共ForkJoinPool。所以你可能需要避免实现又慢又卡的流式操作，因为它可能会拖慢你应用中严重依赖并行流的其它部分。
  51.     }

参考

https://github.com/winterbe/java8-tutorial java8教程

https://wizardforcel.gitbooks.io/modern-java/content/ 中文译站

小二，再来一碟

赏

发表评论

电子邮件地址不会被公开。 必填项已用*标注

6 + 11 = ?