Java函数式编程 - 再谈Stream

1.reduce()

前一章节说了Stream一些使用方式，Stream.reduce()也是Stream中的一个终结操作。使用起来较为复杂一些

1.1 概念

对流中的元素，按照你指定的方式计算出一个结果。reduce的作用是吧stream中的元素组合起来，我们传入一个初始值，他会按照我们计算的方式依次拿流中的元素，在初始化的基础上进行计算，计算结果在和后面的元素计算

关键概念： 初始值的定义（Identity)，累加器（Accumulator)，组合器（Combiner）

Identity : 定义一个元素代表是归并操作的初始值，如果Stream 是空的，也是Stream 的默认结果
Accumulator: 定义一个带两个参数的函数，第一个参数是上个归并函数的返回值，第二个是Strem 中下一个元素
Combiner: 调用一个函数来组合归并操作的结果，当归并是并行执行或者当累加器的函数和累加器的实现类型不匹配时才会调用此函数。

List<Integer> numbers = Arrays.asList(1, 2, 3, 4, 5, 6);int result = numbers.stream().reduce(0, Integer::sum);System.out.println(result);

输出结果：21

说明： 上面示例中，reduce 方法的第一个参数 0 是 identity ，此参数用来保存归并参数的初始值，当Stream 为空时也是默认的返回值。(subtotal, element) -> subtotal + element 是accumulator ，第一个参数是上次累计的和，第二个参数是数据流的下一个元素。如果不传入初始值0会得到一个Optional< Integer > 因为Stream里面的元素可能是0，这样就没法调用reduce了，所以要进一步判断

用reduce 方法处理其他类型的 stream，例如，可以操作一个 String 类型的数组，把数组的字符串进行拼接：

List<String> letters = Arrays.asList("a", "b", "c", "d", "e");String result = letters.stream().reduce("", String ::concat);System.out.println(result);

输出结果：abcde

把字符串转变成大写然后再拼接：

 List<String> letters = Arrays.asList("a", "b", "c", "d", "e");String result = letters.stream().reduce("", (partialString, element) -> partialString.toUpperCase() + element.toUpperCase());System.out.println(result);

输出结果：ABCDE

reduce灵活运用：将配置文件的每一行配置通过map()和reduce()操作聚合成一个Map<String, String>

// 按行读取配置文件:List<String> props = Arrays.asList("profile=native", "debug=true", "logging=warn", "interval=500");Map<String, String> map = props.stream()// 把k=v转换为Map[k]=v:.map(kv -> {String[] ss = kv.split("\\=", 2);Map<String,String> m =new HashMap<>();m.put(ss[0], ss[1]);return m;})// 把所有Map聚合到一个Map:.reduce(new HashMap<String, String>(), (m, kv) -> {m.putAll(kv);return m;});// 打印结果:map.forEach((k, v) -> {System.out.println(k + " = " + v);});

输出结果： logging = warn ，interval = 500，debug = true，profile = native

1.2 其他聚合操作

针对IntStream、LongStream和DoubleStream，还额外提供了以下聚合方法：

sum()：对所有元素求和；
average()：对所有元素求平均数。

2.groupingBy()

分组输出

 List<String> list = Arrays.asList("Apple", "Banana", "Blackberry", "Coconut", "Avocado", "Cherry", "Apricots");Map<String, List<String>> groups = list.stream().collect(Collectors.groupingBy(s -> s.substring(0, 1), Collectors.toList()));System.out.println(groups);

输出结果：{A=[Apple, Avocado, Apricots], B=[Banana, Blackberry], C=[Coconut, Cherry]}

说明： 分组输出使用Collectors.groupingBy()，它需要提供两个函数：一个是分组的key，这里使用s -> s.substring(0, 1)，表示只要首字母相同的String分到一组，第二个是分组的value，这里直接使用Collectors.toList()，表示输出为List

3.文件按行读取

Files类的lines()方法可以把一个文件变成一个Stream，每个元素代表文件的一行内容：

try (Stream<String> lines = Files.lines(Paths.get("/path/to/file.txt"))) {...
}

此方法对于按行遍历文本文件十分有用。

4. stream 基本类型优化

我们之前用到的很多Stream的方法由于都使用了泛型。所以涉及到的参数和返回值都是引用数据类型。
即使我们操作的是整数小数，但是实际用的都是他们的包装类。JDK5中引入的自动装箱和自动拆箱让我们在使用对应的包装类时就好像使用基本数据类型一样方便。但是你一定要知道装箱和拆箱肯定是要消耗时间的。虽然这个时间消耗很下。但是在大量的数据不断的重复装箱拆箱的时候，你就不能无视这个时间损耗了。
所以为了让我们能够对这部分的时间消耗进行优化。Stream还提供了很多专门针对基本数据类型的方法
例如： mapToInt,mapToLong,mapToDouble,flatMapToInt,flatMapDouble

List<Author> authors = getAuthors();authors.stream().map(author -> author.getAge()).map(age -> age + 10).filter(age->age>18).map(age->age+2).forEach(System.out::println);authors.stream().mapToInt(author -> author.getAge()).map(age -> age + 10).filter(age->age>18).map(age->age+2).forEach(System.out::println);

这些就可以避免重复的拆装箱操作，降低性能损耗

5.并行流

当流中有大量的数据的时候，我们可以使用并行流操作提高效率，其实并行流就是把任务分给多个线程去完成，我们自己用代码实现会非常复杂，使用Stream就会比较简单了。

5.1 并行与并发

并行： 多个任务在同一时间点发生，并由不同的cpu进行处理，不互相抢占资源

并发： 多个任务在同一时间点内同时发生了，但由同一个cpu进行处理，互相抢占资源

当在大量数据处理上，数据并行化可以大量缩短任务的执行时间，将一个数据分解成多个部分，然后并行处理，最后将多个结果汇总，得到最终结果。

5.2 基本使用

 Stream<Integer> stream = Stream.of(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);Integer res =//使用并行流 parallel()stream.parallel().filter(s -> s > 5).reduce(Integer::sum).orElseGet(() -> 0);System.out.println(res);

stream.parallel() 调用并行流处理，使用之后后面的操作会由多个线程去处理，我们可以进一步的调试去印证我们的观点，以上代码我做下面的处理，去打印下当前所在的线程

Stream<Integer> stream = Stream.of(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);Integer res =//使用并行流 parallel()stream.parallel().peek(i -> System.out.println(i+","+Thread.currentThread())).filter(s -> s > 5).reduce(Integer::sum).orElseGet(() -> 0);System.out.println(res);

输出结果：

9,Thread[ForkJoinPool.commonPool-worker-2,5,main]
1,Thread[ForkJoinPool.commonPool-worker-6,5,main]
7,Thread[main,5,main]
2,Thread[ForkJoinPool.commonPool-worker-3,5,main]
10,Thread[ForkJoinPool.commonPool-worker-7,5,main]
5,Thread[ForkJoinPool.commonPool-worker-5,5,main]
8,Thread[ForkJoinPool.commonPool-worker-4,5,main]
3,Thread[ForkJoinPool.commonPool-worker-1,5,main]
6,Thread[ForkJoinPool.commonPool-worker-6,5,main]
4,Thread[ForkJoinPool.commonPool-worker-2,5,main]
40

另一种创建并行流的方式：list.parallelStream()

 List<String> list = Arrays.asList("Apple", "Banana", "Blackberry", "Coconut", "Avocado", "Cherry", "Apricots");list.parallelStream() //直接得到一个并行流的对象

注意： 数据量不大的情况下不需要使用并行流