Java8 Stream

「爱情、让人受尽委屈。」 2021-09-10 10:08 411阅读 0赞

## Stream 概述 ##

*  **Stream 是一个可以对序列中的每个元素执行流操作的一个元素序列**
 *  Stream 包含中间和最终两种形式的操作，中间操作返回的还是一个 Stream，因此可以使用链式调用，最终操作返回的是 void 或一个非 Stream 的结果
 *  下面代码中：filter、map、sorted 是中间操作，forEach 是最终操作

List<String> myList =
        Arrays.asList("a1", "a2", "b1", "c2", "c1");
    
    myList
        .stream()
        .filter(s -> s.startsWith("c"))
        .map(String::toUpperCase)
        .sorted()
        .forEach(System.out::println);
    
    // C1
    // C2

*  java Collection 接口中新增了 stream() 方法和 parallelStream() 方法，这些方法可以创建一个顺序的 stream 或者并发的 stream（parallelStream），并发的 stream 适合在多线程中使用

## Stream 创建 ##

*  List 对象调用 stream() 方法返回一个对象流

Arrays.asList("a1", "a2", "a3")
        .stream()
        .findFirst()
        .ifPresent(System.out::println);  // a1

*  也可以使用 Stream.of 方法来创建 Stream

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

*  IntStream、LongStream、DoubleStream 这些流能够处理基本数据类型如：int、long、double
 *  比如：IntStream 可以使用 range() 方法能够替换掉传统的 for 循环

IntStream.range(1, 4)
        .forEach(System.out::println);
    
    // 1
    // 2
    // 3

*  基本类型流比常规对象流类型多了许多聚合方法，如：sum()、average()等

Arrays.stream(new int[] {1, 2, 3})
        .map(n -> 2 * n + 1)
        .average()
        .ifPresent(System.out::println);  // 5.0

*  可以通过常规对象流的 mapToInt() 、mapToLong()、mapToDouble() 方法完成基本类型之间的相互转化

IntStream.range(1, 4)
        .mapToObj(i -> "a" + i)
        .forEach(System.out::println);

*  下面例子 double Stream 先被映射成 int stream，然后又被映射成 String 类型对象流

Stream.of(1.0, 2.0, 3.0)
        .mapToInt(Double::intValue)
        .mapToObj(i -> "a" + i)
        .forEach(System.out::println);
    
    // a1
    // a2
    // a3

## 处理顺序 ##

*  Laziness（延迟加载）是中间操作的一个重要特性，只有最终操作时中间操作才会被执行，如下面例子

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

*  上面代码需要最终操作 forEach 才会执行

Stream.of("d2", "a2", "b1", "b3", "c")
        .filter(s -> {
            System.out.println("filter: " + s);
            return true;
        })
        .forEach(s -> System.out.println("forEach: " + s));
        
    //执行结果如下：
    //filter:  d2
    //forEach: d2
    //filter:  a2
    //forEach: a2
    //filter:  b1
    //forEach: b1
    //filter:  b3
    //forEach: b3
    //filter:  c
    //forEach: c

*  **每一个元素沿着链条垂直移动**，第一个字符串 "d2" 执行完 filter 和 forEach 后才第二个元素，下面代码中，anyMatch 匹配到 A2 时立马返回为真，所以不会继续验证下一个元素 "b1"

Stream.of("d2", "a2", "b1", "b3", "c")
        .map(s -> {
            System.out.println("map: " + s);
            return s.toUpperCase();
        })
        .anyMatch(s -> {最终操作
            System.out.println("anyMatch: " + s);
            return s.startsWith("A");
        });
    
    // map:      d2
    // anyMatch: D2
    // map:      a2
    // anyMatch: A2

## 执行效率与 Stream 执行链顺序的关系 ##

*  下面代码由两个中间操作 map 和 filter 以及一个最终操作 forEach 构成，我们观察这些动作是如何执行的

Stream.of("d2", "a2", "b1", "b3", "c")
        .map(s -> {
            System.out.println("map: " + s);
            return s.toUpperCase();
        })
        .filter(s -> {
            System.out.println("filter: " + s);
            return s.startsWith("A");
        })
        .forEach(s -> System.out.println("forEach: " + s));
    
    // map:     d2
    // filter:  D2
    // map:     a2
    // filter:  A2
    // forEach: A2
    // map:     b1
    // filter:  B1
    // map:     b3
    // filter:  B3
    // map:     c
    // filter:  C

*  上面代码：map 和 filter 被执行了 5 次，但 forEach 只执行了一次，可以修改操作顺序（如：将 filter 操作移动到操作链头部）来降低执行次数
 *  修改后 map 只执行了 1 次，在数据量较大时会有很大的效率提升

Stream.of("d2", "a2", "b1", "b3", "c")
        .filter(s -> {
            System.out.println("filter: " + s);
            return s.startsWith("a");
        })
        .map(s -> {
            System.out.println("map: " + s);
            return s.toUpperCase();
        })
        .forEach(s -> System.out.println("forEach: " + s));
    
    // filter:  d2
    // filter:  a2
    // map:     a2
    // forEach: A2
    // filter:  b1中间操作
    // filter:  b3
    // filter:  c

*  有一些特殊的中间操作不会按流顺序执行，例如 sort 操作，它是一种特殊的中间操作，在对集合元素进行排序的过程中需要保存元素状态，是一种有状态的操作（stateful operation），在执行排序操作时（sorting 操作对集合进行水平操作），上面例子中 sorted 执行了8次，可以通过对链重排序方式，提升 stream 的执行效率

Stream.of("d2", "a2", "b1", "b3", "c")
        .sorted((s1, s2) -> {
            System.out.printf("sort: %s; %s\n", s1, s2);
            return s1.compareTo(s2);
        })
        .filter(s -> {
            System.out.println("filter: " + s);
            return s.startsWith("a");
        })
        .map(s -> {
            System.out.println("map: " + s);
            return s.toUpperCase();
        })
        .forEach(s -> System.out.println("forEach: " + s));
       
    //执行结果如下： 
    //sort:    a2; d2
    //sort:    b1; a2
    //sort:    b1; d2
    //sort:    b1; a2
    //sort:    b3; b1
    //sort:    b3; d2
    //sort:    c; b3
    //sort:    c; d2
    //filter:  a2
    //map:     a2
    //forEach: A2
    //filter:  b1
    //filter:  b3
    //filter:  c
    //filter:  d2

*  下面代码修改链顺序后由于有 filter 操作的过滤，导致 sorted 只有一个输入集元素，在大量数据下能提高效率

Stream.of("d2", "a2", "b1", "b3", "c")
        .filter(s -> {
            System.out.println("filter: " + s);
            return s.startsWith("a");
        })
        .sorted((s1, s2) -> {
            System.out.printf("sort: %s; %s\n", s1, s2);
            return s1.compareTo(s2);
        })
        .map(s -> {
            System.out.println("map: " + s);
            return s.toUpperCase();
        })
        .forEach(s -> System.out.println("forEach: " + s));
    
    // filter:  d2
    // filter:  a2
    // filter:  b1
    // filter:  b3
    // filter:  c
    // map:     a2
    // forEach: A2

## 流复用 ##

*  Stream 当你执行完任何一个最终操作时，流就被关闭了

Stream<String> stream =
        Stream.of("d2", "a2", "b1", "b3", "c")
            .filter(s -> s.startsWith("a"));
    
    stream.anyMatch(s -> true);    // ok
    stream.noneMatch(s -> true);   // exception

*  可以通过 Lambda 表达式创建 Supplier 接口实例，通过 get 方法创建一个 Stream，来实现流复用

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

## Collect（收集） ##

*  假设存在如下用户类

class Person {
        String name;
        int age;
    
        Person(String name, int age) {
            this.name = name;
            this.age = age;
        }
    
        @Override
        public String toString() {
            return name;
        }
    }
    
    List<Person> persons =
        Arrays.asList(
            new Person("Max", 18),
            new Person("Peter", 23),
            new Person("Pamela", 23),
            new Person("David", 12));

*  **Collect 可以把 Stream 中的元素转换为另一种形式**，如：list、set、map
 *  Collect 使用 Collector 作为参数，Collector 包含四种不同操作：
    
     *  supplier（初始构造器）
     *  accaumulator（累加器）
     *  combiner（组合器）
     *  finisher（终结者）
 *  这听起来很复杂，Java8 通过 Collectors 类内置了各种复杂的收集操作，因此对于大部分操作来说不需要自己去实现 Collector 类
 *  例如下面代码将 Stream 转化为 List，如果想转换为 Set 的话，只需要用 Collectors.toSet() 就可以了

Map<Integer, List<Person>> personsByAge = persons
        .stream()
        .collect(Collectors.groupingBy(p -> p.age));
    
    personsByAge
        .forEach((age, p) -> System.out.format("age %s: %s\n", age, p));
    
    // age 18: [Max]
    // age 23: [Peter, Pamela]
    // age 12: [David]

*  下面例子将用户按年龄分组

*  Collectors 类功能繁多，你可以通过 Collectors 对 Stream 元素进行汇聚，如计算平均年龄

Double averageAge = persons
        .stream()
        .collect(Collectors.averagingInt(p -> p.age));
    
    System.out.println(averageAge);     // 19.0

*  可以通过 summarizing collectors 返回一个内置的统计对象，通过这个对象获取更全面的统计信息，比如用户年级中最大值、最小值、平均值等

IntSummaryStatistics ageSummary =
        persons
            .stream()
            .collect(Collectors.summarizingInt(p -> p.age));
    
    System.out.println(ageSummary);
    // IntSummaryStatistics{count=4, sum=76, min=12, average=19.000000, max=23}

*  或者将用户名连接成一个字符串，join collector三个参数分别表示连接符、字符串前缀、字符串后缀

String phrase = persons
        .stream()
        .filter(p -> p.age >= 18)
        .map(p -> p.name)
        .collect(Collectors.joining(" and ", "In Germany ", " are of legal age."));
    
    System.out.println(phrase);
    // In Germany Max and Peter and Pamela are of legal age.

*  可以将 Stream 转换为 map，必须指定 map 的 key 和 value 如何映射，需要注意的是 key 的值必须是唯一的，否则会抛出 IllegalStateException，但可通过合并函数绕过这个异常

Map<Integer, String> map = persons
        .stream()
        .collect(Collectors.toMap(
            p -> p.age,
            p -> p.name,
            (name1, name2) -> name1 + ";" + name2));
    
    System.out.println(map);
    // {18=Max, 23=Peter;Pamela, 12=David}

*  自定义 Collector，实现的功能是将 Stream 中所有用户的用户名变成大写并用 “|” 符号连接成一个字符串，可以通过 Collector.of() 创建一个 collector
 *  由于 String 是不可变对象，所以需要一个辅助类 StringJoiner 帮助 collect 构造字符串，supplier 创建了一个包含分隔符的 StringJoiner 对象，accumulator 将用户名转换为大写并添加到 supplier 创建的 StringJoiner 中，combiner 将两个 StringJoiners 对象连接成一个，最后一步的 finisher 从 StringJoiner 中构建出所希望的 String 对象

Collector<Person, StringJoiner, String> personNameCollector =
        Collector.of(
            () -> new StringJoiner(" | "),          // supplier
            (j, p) -> j.add(p.name.toUpperCase()),  // accumulator
            (j1, j2) -> j1.merge(j2),               // combiner
            StringJoiner::toString);                // finisher
    
    String names = persons
        .stream()
        .collect(personNameCollector);
    
    System.out.println(names);  // MAX | PETER | PAMELA | DAVID

## FlatMap ##

*  虽然我们可以使用 map 方法将 stream 中一种对象转化成另外一种对象，但 map 使用场景有限制，只能将一种对象映射为另外一种已存在的对象，是否能将一个对象映射为多种对象，或者映射成一个根本不存在的对象呢？这就是 flatMap 方法的作用
 *  **FlatMap 方法可以将一个 Stream 中的每个元素转换为另一个 Stream 中的另一种元素**，因此可以将 Stream 种每个对象改造成零、一个或多个
 *  假设有以下类

class Foo {
        String name;
        List<Bar> bars = new ArrayList<>();
    
        Foo(String name) {
            this.name = name;
        }
    }
    
    class Bar {
        String name;
    
        Bar(String name) {
            this.name = name;
        }
    }

*  通过流实例化一队对象

List<Foo> foos = new ArrayList<>();
    
    // create foos
    IntStream
        .range(1, 4)
        .forEach(i -> foos.add(new Foo("Foo" + i)));
    
    // create bars
    foos.forEach(f ->
        IntStream
            .range(1, 4)
            .forEach(i -> f.bars.add(new Bar("Bar" + i + " <- " + f.name))));

*  完成上述操作后我们得到三个 foos，每个 foos 包含三个 bars，下面代码将三个对象的 Stream 转化为一个包含九个对象的 Stream

foos.stream()
        .flatMap(f -> f.bars.stream())
        .forEach(b -> System.out.println(b.name));
    
    // Bar1 <- Foo1
    // Bar2 <- Foo1
    // Bar3 <- Foo1
    // Bar1 <- Foo2
    // Bar2 <- Foo2
    // Bar3 <- Foo2
    // Bar1 <- Foo3
    // Bar2 <- Foo3
    // Bar3 <- Foo3

*  上述代码可以简化成一个管道流

IntStream.range(1, 4)
        .mapToObj(i -> new Foo("Foo" + i))
        .peek(f -> IntStream.range(1, 4)
            .mapToObj(i -> new Bar("Bar" + i + " <- " f.name))
            .forEach(f.bars::add))
        .flatMap(f -> f.bars.stream())
        .forEach(b -> System.out.println(b.name));

## Reduce（减少） ##

*  reduce 操作可以将 Stream 中所有元素组合组合起来得到一个元素，Java8 支持三种不同的 reduce 方法
 *  第一种从 Stream 元素序列中提取一个特定的元素，下面代码从用户列表中选择年纪最大的用户操作，下面代码接收一个二元累加器计算函数

persons
        .stream()
        .reduce((p1, p2) -> p1.age > p2.age ? p1 : p2)
        .ifPresent(System.out::println);    // Pamela

*  第二种 reduce 接收一个标识值和一个二元操作累加器，这个 reduce 方法把 Stream 中所有用户的名字和年龄汇总到一个新的用户

Person result =
        persons
            .stream()
            .reduce(new Person("", 0), (p1, p2) -> {
                p1.age += p2.age;
                p1.name += p2.name;
                return p1;
            });
    
    System.out.format("name=%s; age=%s", result.name, result.age);
    // name=MaxPeterPamelaDavid; age=76

*  第三种 reduce 接收三个参数：一个标识值，一个二元操作累加器，一个二元组合方法

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

## Parallel Streams（并行流） ##

*  为了提高大量输入时的执行效率，Stream 可以采用并行或放行执行，通过 ForkJoinPool.commonPool() 方法获取一个可用的 ForkJoinPool，这个 ForkJoinPool 使用5个线程

ForkJoinPool commonPool = ForkJoinPool.commonPool();
    System.out.println(commonPool.getParallelism());    // 3

*  可通过 jvm 配置修改

-Djava.util.concurrent.ForkJoinPool.common.parallelism=5

*  Collections 中包含 parallelStream() 方法，通过这个方法能够为 Collections 中的元素创建并行流，另外也可以调用 stream 的 parallel() 方法将一个顺序流转换为并行流的拷贝
 *  并行流的执行动作，输出中我们可以看到 parallel stream 使用 ForkJoinPool 提供的可用线程来执行并行流

Arrays.asList("a1", "a2", "b1", "c2", "c1")
        .parallelStream()
        .filter(s -> {
            System.out.format("filter: %s [%s]\n",
                s, Thread.currentThread().getName());
            return true;
        })
        .map(s -> {
            System.out.format("map: %s [%s]\n",
                s, Thread.currentThread().getName());
            return s.toUpperCase();
        })
        .forEach(s -> System.out.format("forEach: %s [%s]\n",
            s, Thread.currentThread().getName()));
    
    //filter:  b1 [main]
    //filter:  a2 [ForkJoinPool.commonPool-worker-1]
    //map:     a2 [ForkJoinPool.commonPool-worker-1]
    //filter:  c2 [ForkJoinPool.commonPool-worker-3]
    //map:     c2 [ForkJoinPool.commonPool-worker-3]
    //filter:  c1 [ForkJoinPool.commonPool-worker-2]
    //map:     c1 [ForkJoinPool.commonPool-worker-2]
    //forEach: C2 [ForkJoinPool.commonPool-worker-3]
    //forEach: A2 [ForkJoinPool.commonPool-worker-1]
    //map:     b1 [main]
    //forEach: B1 [main]
    //filter:  a1 [ForkJoinPool.commonPool-worker-3]
    //map:     a1 [ForkJoinPool.commonPool-worker-3]
    //forEach: A1 [ForkJoinPool.commonPool-worker-3]
    //forEach: C1 [ForkJoinPool.commonPool-worker-2]

*  sort 操作只能在主线程中顺序执行，可以使用 Arrays.parallelSort() 方法来并行执行
 *  parallel streams 可以带来较大的性能提升，但对于 reduce、collect（组合操作） 需要额外的计算，这时并不如顺序流好用

Arrays.asList("a1", "a2", "b1", "c2", "c1")
        .parallelStream()
        .filter(s -> {
            System.out.format("filter: %s [%s]\n",
                s, Thread.currentThread().getName());
            return true;
        })
        .map(s -> {
            System.out.format("map: %s [%s]\n",
                s, Thread.currentThread().getName());
            return s.toUpperCase();
        })
        .sorted((s1, s2) -> {
            System.out.format("sort: %s <> %s [%s]\n",
                s1, s2, Thread.currentThread().getName());
            return s1.compareTo(s2);
        })
        .forEach(s -> System.out.format("forEach: %s [%s]\n",
            s, Thread.currentThread().getName()));
    
    //filter:  c2 [ForkJoinPool.commonPool-worker-3]
    //filter:  c1 [ForkJoinPool.commonPool-worker-2]
    //map:     c1 [ForkJoinPool.commonPool-worker-2]
    //filter:  a2 [ForkJoinPool.commonPool-worker-1]
    //map:     a2 [ForkJoinPool.commonPool-worker-1]
    //filter:  b1 [main]
    //map:     b1 [main]
    //filter:  a1 [ForkJoinPool.commonPool-worker-2]
    //map:     a1 [ForkJoinPool.commonPool-worker-2]
    //map:     c2 [ForkJoinPool.commonPool-worker-3]
    //sort:    A2 <> A1 [main]
    //sort:    B1 <> A2 [main]
    //sort:    C2 <> B1 [main]
    //sort:    C1 <> C2 [main]
    //sort:    C1 <> B1 [main]
    //sort:    C1 <> C2 [main]
    //forEach: A1 [ForkJoinPool.commonPool-worker-1]
    //forEach: C2 [ForkJoinPool.commonPool-worker-3]
    //forEach: B1 [main]
    //forEach: A2 [ForkJoinPool.commonPool-worker-2]
    //forEach: C1 [ForkJoinPool.commonPool-worker-1]

## Stream 常用方法概述： ##

*  **stream**：为集合创建顺序流
 *  **parallelStream**：为集合创建并行流
 *  **forEach**：迭代流中的每个数据
 *  **map**：映射每个元素到对应结果
 *  **collect**：将元素映射为另一种形式

\---- 待完善 ----