终结操作之forEach、forEachOrdered 、reduce、collect

我不是女神ヾ 2022-12-23 12:25 413阅读 0赞

终结操作遍历流来产生一个结果或是副作用。在一个流上执行终结操作之后，该流被消费，无法再次被消费

*  forEach 和 forEachOrdered 对流中的每个元素执行由 Consumer 给定的实现。  
    在使用 forEach时，并没有确定的处理元素的顺序；forEachOrdered 则按照流的相遇顺序来处理元素，如果流有确定的相遇顺序的话
 *  reduce进行递归计算
 *  collect生成新的数据结构

## 1. forEach 和 forEachOrdered ##

//在这里forEach执行是顺序的
    System.out.println("forEach Demo");
    Stream.of("AAA","BBB","CCC").forEach(s->System.out.println("Output:"+s));
    System.out.println("forEachOrdered Demo");
    Stream.of("AAA","BBB","CCC").forEachOrdered(s->System.out.println("Output:"+s));
    
    //输出：
    forEach Demo
    Output:AAA
    Output:BBB
    Output:CCC
    
    forEachOrdered Demo
    Output:AAA
    Output:BBB
    Output:CCC
    
    
    //在这里forEach执行顺序是不能保障的
    Stream.of("AAA","BBB","CCC").parallel().forEach(s->System.out.println("Output:"+s));
    Stream.of("AAA","BBB","CCC").parallel().forEachOrdered(s->System.out.println("Output:"+s));
    
    //第二行将始终输出
    Output:AAA
    Output:BBB
    Output:CCC
    //而第一个不保证，因为forEach处理的元素顺序是不确定的。 
    //forEachOrdered将按其源指定的顺序处理流的元素，无论流是顺序流还是并行流。

## 2. reduce ##

*  reduce一般用于递归操作，最常见的用法就是将stream中一连串的值合成为单个值，比如为一个包含一系列数值的数组求和。
 *  reduce方法有三个重载的方法，方法签名如下：
    
        Optional<T> reduce(BinaryOperator<T> accumulator);
        T reduce(T identity, BinaryOperator<T> accumulator);
        <U> U reduce(U identity,
                         BiFunction<U, ? super T, U> accumulator,
                         BinaryOperator<U> combiner);
 *  reduce执行过程分析
    
        //接受一个BinaryOperator类型的lambada表达式
        List<Integer> numList = Arrays.asList(1,2,3,4,5);
        int result = numList.stream().reduce((a,b) -> a + b ).get();
        System.out.println(result);
    
    代码实现了对numList中的元素累加。lambada表达式的a参数是表达式的执行结果的缓存，也就是表达式这一次的执行结果会被作为下一次执行的参数，而第二个参数b则是依次为stream中每个元素。如果表达式是第一次被执行，a则是stream中的第一个元素。
    
        int result = numList.stream().reduce((a,b) -> { 
          System.out.println("a=" + a + ",b=" + b);
          return a + b;
        } ).get();
    
    在表达式中假如打印参数的代码，打印出来的内容如下：
    
        a=1,b=2
        a=3,b=3
        a=6,b=4
        a=10,b=5
    
        //与第一个签名的实现的唯一区别是它首次执行时表达式第一次参数并不是stream的第一个元素，
        //而是通过签名的第一个参数identity来指定
        //第一种比第一种仅仅多了一个字定义初始值罢了。 
        //-------------------------------------------------------------------
        //此外，因为存在stream为空的情况，所以第一种实现并不直接方法计算的结果，
        //而是将计算结果用Optional来包装，我们可以通过它的get方法获得一个Integer类型的结果
        //，而Integer允许null。
        //
        //第二种实现因为允许指定初始值，因此即使stream为空，也不会出现返回结果为null的情况，
        //当stream为空，reduce为直接把初始值返回
        List<Integer> numList = Arrays.asList(1,2,3,4,5);
        int result = numList.stream().reduce(0,(a,b) ->  a + b );
        System.out.println(result);
    
        //第三种签名的用法相较前两种稍显复杂，由于前两种实现有一个缺陷，
        //它们的计算结果必须和stream中的元素类型相同，如上面的代码示例，stream中的类型为int，
        //那么计算结果也必须为int，这导致了灵活性的不足，甚至无法完成某些任务，
        //比入我们咬对一个一系列int值求和，但是求和的结果用一个int类型已经放不下，
        //必须升级为long类型，此实第三签名就能发挥价值了，它不将执行结果与stream中元素的类型绑死。
        
        List<Integer> numList = Arrays.asList(Integer.MAX_VALUE,Integer.MAX_VALUE);
        long result = numList.stream().reduce(0L,(a,b) ->  a + b, (a,b)-> 0L );
        System.out.println(result);
        
        //如上代码所示，它能将int类型的列表合并成long类型的结果。
        
        //当然这只是其中一种应用罢了，由于摆脱了类型的限制我们还可以通过他来灵活的完成许多任务，
        //比如将一个int类型的ArrayList转换成一个String类型的ArrayList 
        List<Integer> numList = Arrays.asList(1, 2, 3, 4, 5, 6);
        ArrayList<String> result = numList.stream().reduce(new ArrayList<String>(), (a, b) -> { 
            a.add("element-" + Integer.toString(b));
            return a;
        }, (a, b) -> null);
        System.out.println(result);
        
        //执行结果为
        [element-1, element-2, element-3, element-4, element-5, element-6]
        
        //这个示例显得有点鸡肋，一点不实用，不过在这里我们的主要目的是说明代码能达到什么样的效果，
        //因此代码示例也不必取自实际的应用场景。
        
        //现在解释下这个reduce的签名还包含第三个参数，一个BinaryOperator<U>类型的表达式。
        //在常规情况下我们可以忽略这个参数，敷衍了事的随便指定一个表达式即可，目的是为了通过编译器的检查，
        //因为在常规的stream中它并不会被执行到，然而， 虽然此表达式形同虚设，
        //可是我们也不是把它设置为null，否者还是会报错。
        //在并行stream中，此表达式则会被执行到
        /** * lambda语法： * System.out.println(Stream.of(1, 2, 3).parallel().reduce(4, (s1, s2) -> s1 + s2 , (s1, s2) -> s1 + s2)); **/
        System.out.println(Stream.of(1, 2, 3).parallel().reduce(4, 
        		new BiFunction<Integer, Integer, Integer>() { 
        			@Override
        			public Integer apply(Integer integer, Integer integer2) { 
        				return integer + integer2;
        			}
        		}
        		, new BinaryOperator<Integer>() { 
        			@Override
        			public Integer apply(Integer integer, Integer integer2) { 
        				return integer + integer2;
        			}
        		}));
        
        //并行时的计算结果是18，而非并行时的计算结果是10！
        //为什么会这样？
        //先分析下非并行时的计算过程：
        //第一步计算4 + 1 = 5，第二步是5 + 2 = 7，第三步是7 + 3 = 10。这样解释好像没有问题呀。
        //那问题就是非并行的情况与理解有不一致的地方了！
        //先分析下它可能是通过什么方式来并行的？
        //按非并行的方式来看它是分了三步的，每一步都要依赖前一步的运算结果！
        //那应该是没有办法进行并行计算的啊！
        //可实际上现在并行计算出了结果并且关键其结果与非并行时是不一致的！
        //那要不就是理解上有问题，要不就是这种方式在并行计算上存在BUG。
        //暂且认为其不存在BUG，先来看下它是怎么样出这个结果的。
        //猜测初始值4是存储在一个变量result中的；并行计算时，线程之间没有影响，
        //因此每个线程在调用第二个参数BiFunction进行计算时，直接都是使用result值当其第一个参数
        //（由于Stream计算的延迟性，在调用最终方法前，都不会进行实际的运算，
        //因此每个线程取到的result值都是原始的4），因此计算过程现在是这样的：
        //线程1：1 + 4 = 5；线程2：2 + 4 = 6；线程3：3 + 4 = 7；
        //Combiner函数： 5 + 6 + 7 = 18！
        
        //在来一个例子
        /** * lambda语法： * System.out.println(Stream.of(1, 2, 3).parallel().reduce(4, (s1, s2) -> s1 + s2 , (s1, s2) -> s1 * s2)); */
        System.out.println(Stream.of(1, 2, 3).parallel().reduce(4, 
        		new BiFunction<Integer, Integer, Integer>() { 
        			@Override
        			public Integer apply(Integer integer, Integer integer2) { 
        				return integer + integer2;
        			}
        		}
        		, new BinaryOperator<Integer>() { 
        			@Override
        			public Integer apply(Integer integer, Integer integer2) { 
        				return integer * integer2;
        			}
        		}));
        
        //以上示例输出的结果是210！
        //它表示的是，使用4与1、2、3中的所有元素按(s1,s2) -> s1 + s2(accumulator)的方式进行第一次计算，
        //得到结果序列4+1, 4+2, 4+3，即5、6、7；
        //然后将5、6、7按combiner即(s1, s2) -> s1 * s2的方式进行汇总，也就是5 * 6 * 7 = 210。
        //使用函数表示就是：(4+1) * (4+2) * (4+3) = 210;
        //reduce的这种写法可以与以下写法结果相等（但过程是不一样的，三个参数时会进行并行处理）：
        System.out.println(Stream.of(1, 2, 3).map(n -> n + 4).reduce((s1, s2) -> s1 * s2));
        
        //这种方式有助于理解并行三个参数时的场景，实际上就是第一步使用accumulator进行转换
        //（它的两个输入参数一个是identity, 一个是序列中的每一个元素），由N个元素得到N个结果；
        //第二步是使用combiner对第一步的N个结果做汇总。
        /** * 模拟Filter查找其中含有字母a的所有元素，打印结果将是aa ab ad * lambda语法： * s1.parallel().reduce(new ArrayList<String>(), * (r, t) -> {if (predicate.test(t)) r.add(t); return r; }, (r1, r2) -> {System.out.println(r1==r2); return r2; }).stream().forEach(System.out::println); */
        Stream<String> s1 = Stream.of("aa", "ab", "c", "ad");
        Predicate<String> predicate = t -> t.contains("a");
        s1.parallel().reduce(new ArrayList<String>(), 
        	  new BiFunction<ArrayList<String>, String, ArrayList<String>>() { 
        			@Override
        			public ArrayList<String> apply(ArrayList<String> strings, String s) { 
        				if (predicate.test(s)) { 
        					strings.add(s);
        				}
        
        				return strings;
        			}
        		},
        		new BinaryOperator<ArrayList<String>>() { 
        			@Override
        			public ArrayList<String> apply(ArrayList<String> strings, ArrayList<String> strings2) { 
        				System.out.println(strings == strings2);
        				return strings;
        			}
        		}).stream().forEach(System.out::println);
        
          //其中System.out.println(r1==r2)这句打印的结果是什么呢？经过运行后发现是True！
         //为什么会这样？这是因为每次第二个参数也就是accumulator返回的都是第一个参数中New的ArrayList对象！
         //因此combiner中传入的永远都会是这个对象，这样r1与r2就必然是同一样对象！
         //因此如果按理解的，combiner是将不同线程操作的结果汇总起来，那么一般情况下上述代码就会这样写(lambda)：
        
        Stream<String> s1 = Stream.of("aa", "ab", "c", "ad");
        
        //模拟Filter查找其中含有字母a的所有元素，由于使用了r1.addAll(r2)，
        //其打印结果将不会是预期的aa ab ad
        Predicate<String> predicate = t -> t.contains("a");
        s1.parallel().reduce(new ArrayList<String>(), 
        				(r, t) -> { if (predicate.test(t)) r.add(t);  return r; },
        				(r1, r2) -> { r1.addAll(r2); return r1; })
        			.stream().forEach(System.out::println);
        //这个时候出来的结果与预期的结果就完全不一样了，要多了很多元素！
    
        //更多例子
        	// 字符串连接，concat = "ABCD"
        	String concat = Stream.of("A", "B", "C", "D").reduce("", String::concat);
        	// 求最小值，minValue = -3.0
        	double minValue = Stream.of(-1.5, 1.0, -3.0, -2.0).reduce(Double.MAX_VALUE, Double::min);
        	// 求和，sumValue = 10, 有起始值
        	int sumValue = Stream.of(1, 2, 3, 4).reduce(0, Integer::sum);
        	// 求和，sumValue = 10, 无起始值
        	sumValue = Stream.of(1, 2, 3, 4).reduce(Integer::sum).get();
        	// 过滤，字符串连接，concat = "ace"
        	concat = Stream.of("a", "B", "c", "D", "e", "F").
        	 filter(x -> x.compareTo("Z") > 0).
        	 reduce("", String::concat);
        	```

## 3. collect ##

collect含义与Reduce有点相似，先看其定义：

<R> R collect(Supplier<R> supplier,
    			  BiConsumer<R, ? super T> accumulator,
    			  BiConsumer<R, R> combiner);

仍旧先分析其参数（参考其JavaDoc）：

*  supplier：动态的提供初始化的值；创建一个可变的结果容器（JAVADOC）；对于并行计算，这个方法可能被调用多次，每次返回一个新的对象；
 *  accumulator：类型为BiConsumer，注意这个接口是没有返回值的；它必须将一个元素放入结果容器中（JAVADOC）。
 *  combiner：类型也是BiConsumer，因此也没有返回值。它与三参数的Reduce类型，只是在并行计算时汇总不同线程计算的结果。它的输入是两个结果容器，必须将第二个结果容器中的值全部放入第一个结果容器中（JAVADOC）。

可见Collect与分并行与非并行两种情况。  
下面对并行情况进行分析。  
直接使用上面Reduce模拟Filter的示例进行演示(使用lambda语法）：

/** * 模拟Filter查找其中含有字母a的所有元素，打印结果将是aa ab ad */
    Stream<String> s1 = Stream.of("aa", "ab", "c", "ad");
    Predicate<String> predicate = t -> t.contains("a");
    System.out.println(s1.parallel().collect(() -> new ArrayList<String>(),
    		(array, s) -> { if (predicate.test(s)) array.add(s); },
    		(array1, array2) -> array1.addAll(array2)));

根据以上分析，这边理解起来就很容易了：

*  每个线程都创建了一个结果容器ArrayList，假设每个线程处理一个元素，那么处理的结果将会是\[aa\],\[ab\],\[\],\[ad\]四个结果容器（ArrayList）；
 *  最终再调用第三个BiConsumer参数将结果全部Put到第一个List中，因此返回结果就是打印的结果了。

关于collect其他用法：

*  Collectors.toList()：转换成List集合。
 *  Collectors.toSet()：转换成set集合。
 *  Collectors.toCollection()
 *  Collectors.toConcurrentMap()
    
        System.out.println(Stream.of("a", "b", "c","a").collect(Collectors.toSet()));
 *  Collectors.toCollection(TreeSet::new)：转换成特定的set集合。
    
        TreeSet<String> treeSet = Stream.of("a", "c", "b", "a")
        								.collect(Collectors.toCollection(TreeSet::new));
        System.out.println(treeSet);
 *  Collectors.toMap(keyMapper, valueMapper, mergeFunction)：转换成map。
    
        private static void testToConcurrentMap() { 
            Optional.of(menu.stream()
                    .collect(Collectors.toConcurrentMap(Dish::getName, Dish::getCalories)))
                    .ifPresent(v -> { 
                        System.out.println(v);
                        System.out.println(v.getClass());
                    });
        }
        
        private static void testToConcurrentMapWithBinaryOperator() { 
            Optional.of(menu.stream()
                    .collect(Collectors.toConcurrentMap(Dish::getType, v -> 1L, (a, b) -> a + b)))
                    .ifPresent(v -> { 
                        System.out.println(v);
                        System.out.println(v.getClass());
                    });
        }
        
        private static void testToConcurrentMapWithBinaryOperatorAndSupplier() { 
            Map<Thread, StackTraceElement[]> allStackTraces = Thread.getAllStackTraces();
            Optional.of(menu.stream()
                    .collect(Collectors.toConcurrentMap(Dish::getType, v -> 1L, (a, b) -> a + b, ConcurrentSkipListMap::new)))
                    .ifPresent(v -> { 
                        System.out.println(v);
                        System.out.println(v.getClass());
                    });
        }
        ------------------------
        Map<String, String> collect = Stream.of("a", "b", "c", "a")
        									.collect(Collectors.toMap(x -> x, x -> x + x,(oldVal, newVal) -> newVal)));
        collect.forEach((k,v) -> System.out.println(k + ":" + v));
        //a:aa
        //b:bb
        //c:cc
        
        //补充关于合并函数 BinaryOperator<U> mergeFunction对象
        //当toMap中没有mergeFunction时，出现key重复时，会抛出异常 :
        // Exception in thread "main" java.lang.IllegalStateException: Duplicate key aa
        //当使用mergeFunction时,可通过Labmda表达式,对重复值进行处理
 *  Collectors.minBy(Integer::compare)：求最小值，相对应的当然也有maxBy方法。
    
        Optional<String> min = servers.stream.collect(Collectors.minBy(Comparator.comparingInt(String::length)));
 *  Collectors.averagingInt(x->x)：求平均值，同时也有averagingDouble、averagingLong方法。
 *  Collectors.summingInt(x -> x))：求和。
 *  Collectors.summarizingDouble(x -> x)：可以获取最大值、最小值、平均值、总和值、总数。
    
        DoubleSummaryStatistics summaryStatistics = Stream.of(1, 3, 4)
        			.collect(Collectors.summarizingDouble(x -> x));
        System.out.println(summaryStatistics .getAverage());
 *  Collectors.groupingBy(x -> x)：有三种方法，查看源码可以知道前两个方法最终调用第三个方法，  
    第二个参数默认HashMap::new 第三个参数默认Collectors.toList()
    
        //第一种
        Map<Integer, List<Integer>> map = Stream.of(1, 3, 3, 2)
        									.collect(Collectors.groupingBy(Function.identity()));
        System.out.println(map);
        //{1=[1], 2=[2], 3=[3, 3]}
        
        //第二种 
        Map<Integer, Integer> map1 = Stream.of(1, 3, 3, 2)
        		.collect(Collectors.groupingBy(Function.identity(), Collectors.summingInt(x -> x)));
        System.out.println(map1);
        //{1=1, 2=2, 3=6}
        
        //第三种 
        HashMap<Integer, List<Integer>> hashMap = Stream.of(1, 3, 3, 2)
        		.collect(Collectors.groupingBy(Function.identity(), HashMap::new, Collectors.mapping(x -> x + 1, Collectors.toList())));
        System.out.println(hashMap);
        //{1=[2], 2=[3], 3=[4, 4]}
        
        //补充: identity()是Function类的静态方法,和 x->x 是一个意思,
        //当仅仅需要自己返回自己时,使用identity()能更清楚的表达作者的意思.
        //写的复杂一点,绕一点,对理解很有好处
        
        // 按照字符串长度进行分组符合条件的元素将组成一个 List 映射到以条件长度为key的
        //Map<Integer, List<String>> 中
        servers.stream.collect(Collectors.groupingBy(String::length))	
        
        //如果我不想 Map 的 value 为 List 怎么办？ 上面的实现实际上调用了下面的方式：
        //Map<Integer, Set<String>>
        servers.stream.collect(Collectors.groupingBy(String::length, Collectors.toSet()))
        
        //我要考虑同步安全问题怎么办？ 当然使用线程安全的同步容器啊，那前两种都用不成了吧！ 
        //别急！ 我们来推断一下，其实第二种等同于下面的写法: 
        Supplier<Map<Integer,Set<String>>> mapSupplier = HashMap::new;
        Map<Integer,Set<String>> collect = servers.stream
        		.collect(Collectors.groupingBy(String::length, mapSupplier, Collectors.toSet()));
        //这就非常好办了，我们提供一个同步 Map 不就行了，于是问题解决了：
        Supplier<Map<Integer, Set<String>>> mapSupplier = () -> Collections.synchronizedMap(new HashMap<>());
        Map<Integer, Set<String>> collect = servers.stream.collect(Collectors.groupingBy(String::length, mapSupplier, Collectors.toSet()));	
        其实同步安全问题 Collectors 的另一个方法 groupingByConcurrent 给我们提供了解决方案，用法和 groupingBy 差不多。
 *  Collectors.partitioningBy(x -> x > 2)，把数据分成两部分，key为ture/false。第一个方法也是调用第二个方法，第二个参数默认为Collectors.toList()

Map<Boolean, List<Integer>> map = Stream.of(1, 3, 3, 2)
                    .collect(Collectors.partitioningBy(x -> x > 2));
            map.forEach((k,v) -> System.out.println(k + ":" + v));
            
            Map<Boolean, Long> longMap = Stream.of(1, 3, 3, 2)
                    .collect(Collectors.partitioningBy(x -> x > 1, Collectors.counting()));
            longMap.forEach((k,v) -> System.out.println(k + ":" + v));
    		//false:[1, 2]
    		//true:[3, 3]
    		
    		//false:1
    		//true:3

*  Collectors.joining(",")：拼接字符串。
    
        //输出 FelordcnTomcatJettyUndertowResin
         servers.stream().collect(Collectors.joining());
        
         //输出 Felordcn,Tomcat,Jetty,Undertow,Resin
         servers.stream().collect(Collectors.joining("," ));
        
         //输出 [Felordcn,Tomcat,Jetty,Undertow,Resin]
         servers.stream().collect(Collectors.joining(",", "[", "]"));
 *  Collectors.collectingAndThen先执行了一个归纳操作，然后再对归纳的结果进行 Function 函数处理输出一个新的结果
    
        //先执行collect操作后再执行第二个参数的表达式。这里是先塞到集合，再得出集合长度
        Integer integer = Stream.of("1", "2", "3")
        						.collect(Collectors.collectingAndThen(Collectors.toList(), x -> x.size()));
        //输出：3
        
         //我们将servers joining 然后转成大写，结果为： FELORDCN,TOMCAT,JETTY,UNDERTOW,RESIN 
         servers.stream.collect(Collectors.collectingAndThen(Collectors.joining(","), 
         						String::toUpperCase));
 *  Collectors.mapping(…)：跟Stream的map操作类似，只是参数有点区别
    
        //该方法是先对元素使用 Function 进行再加工操作，然后用另一个Collector 归纳。
        //比如我们先去掉 servers 中元素的首字母，然后将它们装入 List 。
        
         // [elordcn, omcat, etty, ndertow, esin]
         servers.stream.collect(Collectors.mapping(s -> s.substring(1), Collectors.toList()));
        
        //有点类似 Stream 先进行了map， 操作再进行collect：
        servers.stream.map(s -> s.substring(1)).collect(Collectors.toList());