Java8 Stream API

朴灿烈づ我的快乐病毒、 2022-05-31 11:59 182阅读 0赞

*  Java8 Stream API
    
     *  Stream是啥
     *  创建流
        
         *  创建一个空的流
         *  通过集合创建
         *  通过数组创建
         *  直接创建
         *  通过builder创建
         *  generate()和iterate()
         *  合并多个Stream
         *  从文件创建
     *  操作流
        
         *  distinct()
         *  filter()
         *  map()与flatMap()
         *  mapToT()与flatMapToT()
         *  limit()
         *  peek()
         *  skip()
         *  sorted()和sorted(Comparator)
     *  收集结果
        
         *  allMatch()
         *  anyMatch()
         *  collect(collector)和collect(supplier, accumulator, combiner)
         *  count()
         *  findAny()和findFirst()
         *  forEach()和forEachOrdered()
         *  max()和min()
         *  noneMatch()
         *  reduce()
         *  toArray()和toArray(generator)
     *  IntegerStream、DoubleStream、LongStream
     *  一些使用中遇到的问题
        
         *  Exception
         *  Collectors.toMap()

# Java8 Stream API #

## Stream是啥 ##

*  用一段时间`Stream API`之后，会发现“流”这个称呼非常的贴切，不抄书上的解释的话，流很像一个流水线：先把集合(暂时忽略`IntStream`等)拆成一个一个的放到一个流水线上，然后在流水线上有很多工人或机械臂(比如筛选、重新映射、去重等等)，最后在流水线末端有一个收集装置(比如重新收集成集合、取出最大值、分组等等)，将产品包装成我们想要的样子。
 *  所以一个`Stream API`的使用过程一般分为三个步骤：创建流->操作流->收集结果，这次学习笔记主要从这三个方面记录和完善。
 *  主线只针对集合的流式操作，`IntStream` `LongStream` `DoubleStream`单独学习并在一个独立的模块中记录。
 *  需要先了解`lambda expression`。

## 创建流 ##

### 创建一个空的流 ###

正常情况下不会创建一个空的流，一般用来预防`NPE`.

public Stream streamOf(Collection collection){
        if(collection != null && !collection.isEmpty()){
            return collection.stream();
        }
        return Stream.empty();
    }

### 通过集合创建 ###

`Collection`接口有一个`stream()`方法并且有`default`实现，任何继承自`Collection`的类都能直接创建流。

//Collection.class
    default Stream<E> stream() {
        return StreamSupport.stream(spliterator(), false);
    }

List<String> list = new ArrayList<>();
    Stream stream = list.stream();

### 通过数组创建 ###

`Arrays.stream()`有很多重载方法，可以按需使用。

String[] array = new String[]{
       "A","B","C"};
    Stream stream = Arrays.stream(array);

### 直接创建 ###

`Stream.of()`方法参数是可变长的。

Stream stream = Stream.of("A","B","C");

### 通过builder创建 ###

Stream stream = Stream.builder()
            .add("A")
            .add("B")
            .add("C")
            .build();

### generate()和iterate() ###

两个都是生成一个无限的流，通常跟`limit()`一起使用，限制流中元素的个数。不同的是前者可以根据任何计算方式来生成，后者只能根据给定的`seed`来生成，自我感觉这个两个方法在处理一些数学公式或时非常实用，下面的例子用`generate()`打印前10个斐波那契数列项。

Stream stream = Stream.generate(new Supplier<Long>() {
        long a = 0,b = 1;
        @Override
        public Long get() {
            long tmp = a + b;
            a = b;
            b = tmp;
            return a;
        }
    });
    stream.limit(10).forEach(System.out::println);

接下来是用`itrate()`打印20\-210。`itrate()`生成的元素与`seed`(第一个参数)密切相关，相当于是`f(seed)`、`f(f(seed))`、`f(f(f(seed)))`……

Stream stream = Stream.iterate(1, n -> n * 2);
    stream.limit(11).forEach(System.out::println);

### 合并多个Stream ###

Stream stream1 = Stream.builder()
             .add("A")
             .add("B")
             .add("C")
             .build();
     Stream stream2 = Stream.builder()
             .add("D")
             .add("E")
             .add("F")
             .build();
     Stream stream = Stream.concat(stream1,stream2);

### 从文件创建 ###

try(Stream<String> stream = Files.lines(Paths.get("C:\\Windows\\System32\\drivers\\etc\\hosts"), Charset.defaultCharset())){
        stream.forEach(System.out::println);
    }catch (IOException e){
        e.printStackTrace();
    }

## 操作流 ##

操作流的结果依然是一个流，就像是从一个分区转移到另一个分区一样，不到收集结果阶段，所有元素都依然在流水线上生存，只是不同分区有不同的功能而已。

### distinct() ###

去重，把流水线上相同的元素去掉，只保留不同的元素。

Stream.of(1,2,1).distinct().forEach(System.out::println);//1,2

### filter() ###

过滤，把满足指定条件的元素留在流水线上，其他的删掉。

Stream.of(10,3,9,5).filter(n -> n > 5).forEach(System.out::println);//10,9

### map()与flatMap() ###

`map()`是把流水线上的产品挨个挨个做相同的处理，比如给每个产品贴个标签，每个数字`+1`；`flatMap`更像是拆箱，放到流水线上的产品是被箱子包装起来的，先要把箱子拆开把里面的产品放到流水线上再做后续处理。`map()`是直接对流水线上的产品做处理，即使有“箱子”也会被忽略，标签会直接贴在箱子上；`flatMap()`目的是对箱子里的产品做处理。因此`map()`的参数是具体操作，而`flatMap()`的参数是一个`Stream`，即`Stream`就是箱子。

Stream.of(new ArrayList<>(Arrays.asList(1,2,3)),new ArrayList<>(Arrays.asList(10,20,30)))
          .map(item -> item.subList(0,1))
          .collect(Collectors.toList())
          .forEach(System.out::println);//[1] [10]
    Stream.of(new ArrayList<>(Arrays.asList(1,2,3)),new ArrayList<>(Arrays.asList(10,20,30)))
          .flatMap(item -> item.stream())
          .map(item -> item + 1)
          .collect(Collectors.toList())
          .forEach(System.out::println);//2 3 4 11 21 31

### mapToT()与flatMapToT() ###

这两类方法包括`flatMapToDouble()` `flatMapToInt()` `flatMapToLong()`以及`mapToDouble()` `mapToInt()` `mapToLong()`，功能大体上和`map()`与`flatMap()`相同，只不过针对的产品不同：对于`Double`型的产品可以放到`DoubleStream`流水线上处理，这个流水线上可能包含新的功能区，当然产品放到`DoubleStream`流水线前，必须保证产品是`Double`类型的(参数的返回值必须是`Double`类型)。`DoubleStream`会单独记录，现在只考虑如何放到流水线上，不考虑流水线的任何功能与操作。

Stream.of(1,2,3).mapToDouble(Double::new).forEach(System.out::println);

### limit() ###

限制产品线上的产品数量，不超过指定的数量。

Stream.of(1,2,3).limit(1).forEach(System.out::println);//1
    Stream.of(1,2,3).limit(10).forEach(System.out::println);//1 2 3

### peek() ###

给产品安装一个监听器，当产品下线被收集时，将触发所有的监听器，这个监听器可以拿到产品当时的状态，注意是当时的状态哦，不是最终的状态，然后就可以在这个监听器里为所欲为了。因为流水线上的产品只能被消费一次，因此监听器只会被触发一次，不可能多次被触发。

Stream.of(1,2,3)
        .peek(item -> System.out.println("consumer1 [" + item + "]"))
        .map(item -> item + 1)
        .peek(item -> System.out.println("consumer2 [" + item + "]"));
    //没有消费(收集)过程，输出空
    //因此不消费是不会触发peek

Stream.of(1,2,3)
        .peek(item -> System.out.println("consumer1 [" + item + "]"))
        .map(item -> item + 1)
        .peek(item -> System.out.println("consumer2 [" + item + "]"))
        .forEach(System.out::println);
    //最终输出
    /* consumer1 [1]//第一个peek()的时候还没+1 consumer2 [2]//第一个peek()的时候已经+1，因此是当时的状态 2 //是在正真消费之前触发的 consumer1 [2] consumer2 [3] 3 consumer1 [3] consumer2 [4] 4 */

可见每一次`peek()`都是存了快照的，`Java API`文档里都说了：可以利用这个特性来做调试。

### skip() ###

和`limit()`恰好相反，`skip()`是跳过流水线上前`n`个产品，保留剩下的产品。

Stream.of(1,2,3).skip(2).forEach(System.out::println);//3
    Stream.of(1,2,3).skip(5).forEach(System.out::println);//空

### sorted()和sorted(Comparator) ###

明显，前者是根据元素自然排序，后者是根据指定的策略排序。自定义的排序规则可以调用`Comparator`的静态方法，也可以自己写。`Comparator`的静态方法几乎都是按照自然排序排的，即使是自己写的比较器，也是“小的”放在前面，可以使用`reverseOrder()`和`reversed()`反序。

Stream.of("5","7","0","a","z","^").sorted().forEach(System.out::println);//0 5 7 ^ a z
    Stream.of("4444","22","333","1","55555").sorted(Comparator.comparingInt(String::length).reversed()).forEach(System.out::println);//按字符串长度倒序排
    Stream.of("4444","22","333","1","55555").sorted((a,b) -> b.length() - a.length()).forEach(System.out::println);//跟上面的效果一样

## 收集结果 ##

### allMatch() ###

返回流中的元素是否全部满足给定的条件，相当有用。

List<Integer> list = Arrays.asList(1,2,3,4,5);
    System.out.println(list.stream().allMatch(s-> s > 0));//true
    System.out.println(list.stream().allMatch(s-> s > 1));//false

### anyMatch() ###

返回流中的元素是否有任意一个满足给定的条件，也很有用的。

List<Integer> list = Arrays.asList(1,2,3,4,5);
    System.out.println(list.stream().anyMatch(s-> s > 4));//true
    System.out.println(list.stream().anyMatch(s-> s > 10));//false

### collect(collector)和collect(supplier, accumulator, combiner) ###

用`Collector`来收集结果，包括转换成各种集合、总数、求和、求均值、分组、分区等等。

System.out.println(Stream.of(4444,22,333,1,55555)
                    .collect(Collectors.summarizingInt(item -> item)));
    //输出：IntSummaryStatistics{count=5, sum=60355, min=1, average=12071.000000, max=55555}
    System.out.println(Stream.of(4444, 22, 333, 1, 55555)
                    .collect((Supplier<ArrayList>) ArrayList::new, ArrayList::add, ArrayList::addAll));
    //输出：[4444, 22, 333, 1, 55555]
    System.out.println(Stream.of(4444, 22, 333, 1, 55555)
                    .collect(Collectors.toList()));//跟上面一样

### count() ###

返回流中元素个数.

System.out.println(Stream.of(4444, 22, 333, 1, 55555).count());//5

### findAny()和findFirst() ###

这两个方法其实是一样的，`findAny() java doc`这样写的：

> The behavior of this operation is explicitly nondeterministic; it is free to select any element in the stream. This is to allow for maximal performance in parallel operations; the cost is that multiple invocations on the same source may not return the same result. (If a stable result is desired, use findFirst() instead.)

看起来是说`findAny()`是返回任意一个元素，但是实际情况并不是这样：

Stream.of(4444, 22, 333, 1, 55555).findFirst().ifPresent(System.out::println);//4444
    Stream.of(4444, 22, 333, 1, 55555).findAny().ifPresent(System.out::println);//4444
    Stream.of(4444, 22, 333, 1, 55555).findAny().ifPresent(System.out::println);//4444
    Stream.of(4444, 22, 333, 1, 55555).findAny().ifPresent(System.out::println);//4444
    Stream.of(4444, 22, 333, 1, 55555).findAny().ifPresent(System.out::println);//4444
    Stream.of(4444, 22, 333, 1, 55555).findAny().ifPresent(System.out::println);//4444
    Stream.of(4444, 22, 333, 1, 55555).findAny().ifPresent(System.out::println);//4444

这里有一个解释：[Java 8 Stream.findAny() vs finding a random element in the stream  
][Java 8 Stream.findAny_ vs finding a random element in the stream]  
结合`java doc`和`stackoverflow`上的第一个回答翻译过来就是：

> `findAny()`实际上是`findFirst()`另一个更灵活的选择，在某些情况下（并行流操作）`findAny()`的开销更少，但是代价是同一个数据源多次调用`findAny()`可能结果不一样。简单的说就是：`findFirst()`一定是第一个元素，`findAny()`能取出某个元素，但不保证是第一个，也不能保证每次取到是同一个。

原来只是在并行流(`parallel stream`)的时候，两个方法才有区别的。

List list = Arrays.asList(4444, 22, 333, 1, 55555);
    IntStream.iterate(0,i -> i + 1).limit(10).forEach(i -> list.parallelStream().findAny().ifPresent(System.out::println));
    //多运行几次可能会输出：
    /* 333 333 333 333 333 333 22 333 22 333 */
    //但是findFirst()无论如何都是第一个
    //并行遍历时查找第一个应该需要更多的代价，findAny()可以用更少的代价从流中去取一个元素，而且也没有明显的随机效果
    //在列表中所有元素等价并且是并行流的时候，用findAny()开销比findFirst()低，其他情况还是findFirst()吧，稳一些
    //注意：两个方法中的任意一个方法、在任何情况下都没有很好的随机效果

### forEach()和forEachOrdered() ###

这两个方法就是遍历，前面用了好多次，可以每个元素调用一个方法，比如打印。`forEachOrdered()`与`forEach()`的关系和`findAny()`与`findFirst()`的关系相似，前者是在并行流的情况下依然按输入顺序遍历，当然单价是更大的开销。

List list = Arrays.asList(1, 2, 3);
    IntStream.iterate(0,i -> i + 1).limit(3).forEach(i -> {
        synchronized (StreamTeat.class){
            list.parallelStream().forEach(System.out::print);
            System.out.println();
        }
    });
    //可能的输出
    /* 123 321 213 */
    //即并行遍历的时候输出顺序是不定的，如果用forEachOrdered()那么肯定是按照输入顺序遍历的

### max()和min() ###

跟`sorted()`方法结合起来看，`Comparator`是必须的。

Stream.of(4444, 22, 333, 1, 55555).max(Comparator.naturalOrder()).ifPresent(System.out::println);//55555
    Stream.of(4444, 22, 333, 1, 55555).min(Comparator.naturalOrder()).ifPresent(System.out::println);//1

### noneMatch() ###

和`anyMatch()`相反。

System.out.println(Stream.of(4444, 22, 333, 1, 55555).noneMatch(item -> item < 0));//true
    System.out.println(Stream.of(4444, 22, 333, 1, 55555).noneMatch(item -> item > 3));//false

### reduce() ###

规约，把流中的元素前两个执行一个方法，再把结果和第三个元素执行同样的方法，直至最后一个元素，最后得出结果：可以定义初始值，也可以定义返回类型和规约操作，比如可以用规约实现一个`sum()`，和`collect()`的`Collectors.reduce()`很像的，有三个重载方法。

System.out.println(Stream.of(1, 2, 3, 4).reduce(((sum,item) -> sum += item)));//10
    System.out.println(Stream.of(1, 2, 3, 4).reduce(656,((sum,item) -> sum += item)));//666
    System.out.println(Stream.of(1, 2, 3, 4).parallel().reduce(new StringBuilder(), StringBuilder::append, StringBuilder::append));//1234
    //第一个方法：把流中的元素前两个执行一个方法，再把结果和第三个元素执行同样的方法，直至最后一个元素，返回类型和元素类型一致
    //第二个方法：把初始值656和流中的第一个元素执行一个方法，再把结果和第二个元素执行同样的方法，直至最后一个元素，返回类型和元素类型一致
    //第三个方法：定义返回类型，定义规约操作，定义并行流结果合并方式

第三个方法可以参开这里：  
[java8中3个参数的reduce方法怎么理解？][java8_3_reduce]

> 意思就是并行的时候，流被分成多段，每段会产生一个同样类型的结果，比如有100个产品在流水线上，被分配给10个工人，最终要装在盒子里；10个工人每个人都会把自己的10个产品装在一个盒子里，最终这10个盒子要被合并在一个盒子里，那么盒子与盒子之间要定义合并规则，所以第三个参数在并行流的时候才会用到。

**注意：并行流时第三个参数可能有重复元素，这里没有做太深入的了解，应该需要注意排重**

### toArray()和toArray(generator) ###

都能返回一个流中所有元素组成的`array`，后者可以有自定义数组元素类型。

System.out.println(Arrays.toString(Stream.of(1, 2, 3, 4).toArray()));//[1, 2, 3, 4]
    System.out.println(Arrays.toString(Stream.of(1, 2, 3, 4).toArray(Integer[]::new)));//[1, 2, 3, 4]
    System.out.println(Arrays.toString(Stream.of(1, 2, 3, 4).toArray(size -> new Integer[size])));//上面的方法就是把数组的size传进来了
    //第一个方法是返回Object[]，第二个方法是返回Integer[]

## IntegerStream、DoubleStream、LongStream ##

全都是`Stream`的一些特殊实现。

*  约束性更强，元素类型固定，一些特殊方法比如`summaryStatistics()`可以直接调用，而不用在`Collect()`里面才能调用。
 *  一些新的方法比如`range()`和`rangeClosed()`方法来生成一个流，类似于`fork i++`和`fork i--`。
 *  可以用`IntStream.mapToObj()`转换成`Stream`；同样可以用`Stream.mapToInt`转换成`Stream`。

## 一些使用中遇到的问题 ##

### Exception ###

因为`lambda`表达式和匿名内部类有些相似，可以看做一个闭包，`Exception`必须在内部`catch`住而不能`throw`出来让外层处理，所以在`Stream`中的`lambda`表达式调用一个声明`throw Exception`的方法时很不友好，直接编译错误。这样就不能让外层中断，外层甚至不能轻易地获取错误(可以用一个全局变量保存错误，但只有循环完毕才能拿到错误信息，可以再给这个全局变量加个监听，保证出错时能第一时间获取错误)，这时就不必强行使用`Stream`了。  
还有一种解决方案是，让方法抛出`RuntimeException`，编译肯定能通过，内层出错也能立即终止，但是如果方法无法更改，那也无能为力。

### Collectors.toMap() ###

这个方法有点坑的，如果某个`value`为`null`，会报错的。因为`toMap()`方法虽然有三个重载方法，但是都没有包含所有的参数，底层的`java.util.stream.Collectors.CollectorImpl`构造函数是有5个参数的，其中有个`BinaryOperator<A> combiner`参数，这个方法是用来解决`key`冲突的，默认会调用`Map.merge()`方法，但对用户不可见，无法直接传入，这个方法要求`value`不能为空，否者报`NPE`。  
这个其实也有解决方法的，因为`merge()`不是用来解决`key`冲突的嘛，自己写个类实现`java.util.stream.Collector`把`combiner`开放出来就好了。

System.out.println(Stream.of("1","2","3",null).collect(Collectors.toMap(k -> k,v -> v)));
    //Exception in thread "main" java.lang.NullPointerException
    
    System.out.println(Stream.of("1","2","3","3").collect(Collectors.toMap(k -> k,v -> v)));
    //Exception in thread "main" java.lang.IllegalStateException: Duplicate key 3
    //默认解决冲突的方法是直接抛出错误

自己写个`toMap()`方法解决这个问题。

public class DbaasCollectors { 
    
        static class ToMapCollector<T,K,V> implements Collector<T,Map<K,V>,Map<K,V>>{
    
            private Function<? super T, ? extends K> keyMapper;
    
            private Function<? super T, ? extends V> valueMapper;
    
            private BinaryOperator<Map<K, V>> combiner;
    
            public ToMapCollector(Function<? super T, ? extends K> keyMapper,
                                  Function<? super T, ? extends V> valueMapper,
                                  BinaryOperator<Map<K, V>> combiner) {
                super();
                this.keyMapper = keyMapper;
                this.valueMapper = valueMapper;
                this.combiner = combiner;
            }
    
            @Override
            public Supplier<Map<K, V>> supplier() {
                return HashMap::new;
            }
    
            @Override
            public BiConsumer<Map<K, V>, T> accumulator() {
                return (map, element) -> map.put(keyMapper.apply(element), valueMapper.apply(element));
            }
    
            @Override
            public BinaryOperator<Map<K, V>> combiner() {
                return combiner;
            }
    
            @Override
            public Function<Map<K, V>, Map<K, V>> finisher() {
                return (kvMap -> (Map<K, V>) kvMap);
            }
    
            @Override
            public Set<Characteristics> characteristics() {
                return Collections.unmodifiableSet(EnumSet.of(Collector.Characteristics.IDENTITY_FINISH));
            }
        }
    
        public static <T, K, V> Collector<T, ?, Map<K, V>> toMap(Function<T, K> keyMapper, Function<T, V> valueMapper, BinaryOperator<Map<K, V>> combiner) {
            return new ToMapCollector<>(keyMapper,valueMapper,combiner);
        }
    
        private static <K, V> Map<K, V> merge(Map<K, V> result1, Map<K, V> result2) {
            result2.forEach((key, value) -> {
                if (result1.containsKey(key)) {
                    result1.put(key, (V) (String.valueOf(result1.get(key)) + String.valueOf(value)));
                } else {
                    result1.put(key, value);
                }
            });
            return result1;
        }
    
        public static void main(String[] args){
            System.out.println(Stream.of("1","2","3","3",null).parallel().collect(toMap(k -> k, v -> v, DbaasCollectors::merge)));
        }
    }
    //输出：{null=null, 1=1, 2=2, 3=33}
    //既解决了value不能为空的问题，又可以自定义merge方法，还解决了key重复报错的问题

[Java 8 Stream.findAny_ vs finding a random element in the stream]: https://stackoverflow.com/questions/25912185/java-8-stream-findany-vs-finding-a-random-element-in-the-stream
[java8_3_reduce]: https://segmentfault.com/q/1010000004944450