流式计算-Java8Stream

Posted 2021-03-22 甲由崽

大约各位看官君多少也听说了Storm/Spark/Flink，这些都是大数据流式处理框架。如果一条手机组装流水线上不同的人做不同的事，有的装电池，有的装屏幕，直到最后完成，这就是典型的流式处理。如果手机组装是先全部装完电池，再交给装屏幕的组，直到完成，这就是旧式的集合式处理。今天，就来先说说JDK8中的流，虽然不是很个特新鲜的话题，但是一个很好的开始，因为——思想往往比细节重要！

 

作者原创文章，谢绝一切转载！

 本文只发表在"公众号"和"博客园"，其他均属复制粘贴！如果觉得排版不清晰，请查看公众号文章。 

 

准备：

Idea2019.03/Gradle5.6.2/JDK11.0.4/Lambda

难度：新手--战士--老兵--大师

目标：

1.Lambda表达式使用

2.流的筛选/切片/查找/匹配/映射/归约操作

步骤：

为了遇见各种问题，同时保持时效性，我尽量使用最新的软件版本。代码地址：其中的day22，https://github.com/xiexiaobiao/dubbo-project.git

1.先来两个概念：

• 流(Stream)：一个元素序列。位于包java.util.stream.Stream，注意这个序列是可以串行或并行处理的。有多种方式建立流，最常见的是从集合(Collection)对象获取，有序集合如List的流有序，Set的流则无序。
• Lambda表达式：流式处理的绝佳搭档！什么是Lambda表达式？略。哪里可以用Lambda表达式？需要实现一个函数式接口(只定义了一个抽象函数的接口)的地方就可以使用Lambda表达式，代替匿名类方式。源代码中com.biao.lambda包里，我写了一个简单的Lambda实例，供参考。

2.流式处理特点：

• 流水线：流操作可返回一个流，多个操作从而可形成一个链，
• 内部迭代：使用Iterator/forEach显式迭代器为外部迭代，流的迭代是流内部完成的，只需声明，是内部迭代，
• 一次使用：每个流只能消费一次，不能结束后又从头开始！

3.流的一般使用：

• 建立流：创建一个Stream对象，如从一个数据源来执行一个查询；

• 操作流：一个包含了各种操作的操作链；

• 结束流：一个终端操作，形成一个结果集或值

 

4.来个例子，假设这里有个com.biao.Fruit类：

@Data
public class Fruit {
    private String name;
    private String origin;
    private Integer price;
}

我们要从一堆水果里，找前4种产自中国的名称是字母A开头的水果。这还不是小菜？几次使用Iterator或者forEach循环就实现了！

如果使用流处理，大致处理流程图示则如下，代码后面再聊：

 

5.流的主要操作：筛选/切片/查找/匹配/映射/归约

下面将一一道来：

筛选/切片：使用filter/skip/limit/distinct方法过滤。filter接收一个Predicate函数表达式，方法签名是T --> boolean，我们来实现上面的图示逻辑，至于JDK7的实现，看官君可以想一想，对比一下，com.biao.Application1代码实现片段：

// 创建流
        Stream<Fruit> fruitStream = fruitList.stream();
        // 过滤
        Stream<Fruit> filteredStream = fruitStream.filter(d -> "China".equals(d.getOrigin()));
        // 去掉重复元素
        Stream<Fruit> distinctStream = filteredStream.distinct();
        // 打印流中元素，forEach是终端操作，如果这里使用了，则collect方法无法使用,即一个流只能消费一次
        // distinctStream.forEach(System.out::println);
        // 跳过1个元素，
        Stream<Fruit> skippedStream = distinctStream.skip(1);
        // 切片,参数为maxSize
        Stream<Fruit> limitStream = skippedStream.limit(4);
        // 结束，collect方法是收集器，如果这里使用了，则forEach无法使用，即一个流只能有一个终端操作
        List<Fruit> newList = limitStream.collect(Collectors.toList());
        // 打印结果，lambda方式
        newList.forEach(System.out::println);

        // 链式操作,和上面效果一样，一气呵成，真爽！
        List<Fruit> newList2 = fruitList.stream()
                .filter(d -> "China".equals(d.getOrigin()))
                .distinct()
                .skip(1)
                .limit(4)
                .collect(Collectors.toList());
        // 打印结果集
        newList2.forEach(System.out::println);

以上代码核心点：

• 尽量使用链式语法配合Lambda，简洁至极！
• 一个流只能有一个终端操作！即一个流只能被消费一次！
• filter方法的参数表达式可以进行逻辑复合运算，如and/not/or，

映射：对流中的每个元素应用映射函数，变换成新的对象。使用map方法，接受一个Function类型，函数签名是 T—> R，比如对以上Fruit流提取水果的名称，并过滤字母A开头的水果，com.biao.Application2代码实现片段：

// 创建流
Stream<Fruit> fruitStream = fruitList.stream();
//转换,变为String流
Stream<String> stringStream = fruitStream.map(Fruit::getName);
//过滤，名称以A开头
Stream<String> filteredStream = stringStream.filter(str -> "A".equals(String.valueOf(str.charAt(0))));
//终端操作,set自动去重复
Set<String> stringSet = filteredStream.collect(Collectors.toSet());
//打印结果集
stringSet.forEach(System.out::println);

//链式语法实现,请君想象下JDK7的实现，
fruitList.stream()
      .map(Fruit::getName)
      .filter(str -> "A".equals(str.substring(0,1)))
      .collect(Collectors.toSet())
      .forEach(System.out::println);

 

我还写了个map映射+flatMap扁平化流例子，com.biao.Application3代码片段：

 /**映射示例2：map映射+flatMap扁平化流*/
String[] arraysOfWords = {"Apple","Banana","Nuts","Olive"};
// 使用Arrays的静态方法创建流
Stream<String> stringStream = Arrays.stream(arraysOfWords);
// 对每个word映射为String[]
stringStream.map(word -> word.split(""))
      // flatMap扁平化流，将生成的流组合成一个流
      // 如果使用map(Arrays::stream),则生成由流元素组成的流
      .flatMap(Arrays::stream)
      // 去掉重复
      .distinct()
      // 终端操作，collect方法是收集器
      .collect(Collectors.toList())
      .forEach(System.out::println);

 

流的扁平化，一言以蔽之，flatmap方法让你把一个流中的每个值都换成一个的流(即流中的元素也是流)，然后把所有的流连接起来成为一个流。

查找/匹配：StreamAPI通过allMatch,anyMatch,noneMatch,findFirst,findAny方法找到符合的元素，com.biao.Application4代码实现片段：

   // 注意这里每个都要重建一个流
        // 是否全部价格大于50
        boolean almach = fruitList.stream().allMatch(fruit -> fruit.getPrice() > 50);
        System.out.println(almach);
        // 是否至少有一种产自America
        boolean anyMatch = fruitList.stream().anyMatch(fruit -> "America".equals(fruit.getOrigin()));
        System.out.println(anyMatch);
        // 找出流中第3个，
        Optional<Fruit> thirdOne = fruitList.stream().skip(2).findFirst();
        // 存在则打印，防止NPE
        thirdOne.ifPresent(System.out::println);
        // 找出流中任意一个，，
        Optional<Fruit> anyOne = fruitList.stream().findAny();
        // ifPresent，值存在则执行操作，否则 do nothing！
        anyOne.ifPresent(System.out::println);

 

以上代码核心点：

• 这里每个查找/匹配都要重建一个流，
• 找可能的第3个元素，skip(2).findFirst()，返回Optional T 类，解决返回null的NPE问题，这样即使不存在第3个元素，返回对象仍然可以继续做计算，
• findAny时，流水线将在后台进行优化使其只需走一遍，并在利用短路找到结果时立即结束

归约：使用reduce对流中元素累积计算，最后得到一个值。比如找到上面水果中价格最高的，计算出产自Japan的水果的总价格，com.biao.Application5代码实现片段：

// 注意这里每个都要重建一个流
        int totalPrice = fruitList.stream()
                .filter(fruit -> "Japan".equals(fruit.getOrigin()))
                //映射转换为Integer流
                .map(Fruit::getPrice)
                //reduce归约计算
                // 也可使用reduce(0,(a,b) -> a+b);
                .reduce(0,Integer::sum);
        System.out.println(totalPrice);

        /** reduce无初始值的归约计算 */
        Optional<Integer> totalPrice2 = fruitList.stream()
                .map(Fruit::getPrice)
                .reduce((a,b) -> a+b);
        // ifPresent，值存在则执行操作，否则 do nothing！
        totalPrice2.ifPresent(System.out::println);

        /** reduce计算最大*/
        Optional<Integer> maxPrice = fruitList.stream()
                .map(Fruit::getPrice)
                // 归约计算最大值：
                // 这里也可以使用reduce((x,y) -> x>y?x:y)
                .reduce(Integer::max);
        // ifPresent，值存在则执行操作，否则 do nothing！
        maxPrice.ifPresent(System.out::println);

        /** reduce计算最小值*/
        Optional<Integer> minPrice = fruitList.stream()
                .map(Fruit::getPrice)
                // 归约计算最小值：也可以使用reduce((x,y) -> x<y?x:y)
                .reduce(Integer::min);
        // ifPresent，值存在则执行操作，否则 do nothing！
        minPrice.ifPresent(System.out::println);

 

以上代码核心点：

• reduce的函数参数有几种重载，返回的值不一样，无初始值的返回Optional对象，
• Optional. ifPresent方法，Optional对象值存在则执行操作，否则 do nothing，
• map和reduce的连接通常称为map-reduce模式，源于google的搜索模式，

6.看完了上面的各种流操作，看官君也许会说，似乎也没啥大不了的啊，顶多是少写了几行代码，那请用集合式实现下相同逻辑！另请君回忆下，是否有过多层嵌套或者N多分支的if-elseif-else/switch-case场景？那现在就请试试流式写法！事实上我这里只是举了几个常规的应用例子而已，抓住下看官君的兴趣，StreamAPI还有其他强大的功能：

• 无限流、范围流：能直接创建无限流，再局部处理，
• Collect收集器的分区分组：将最终结果集按条件分区分组，类比SQL的groupBy，
• 流的parallel/sequential并行计算和顺序计算：声明式并行计算，无需为锁烦恼，
• 分支/合并ForkJoin框架的递归计算：多线程方式处理，还可自定义线程池参数，
• 同步/异步执行：使用CompletableFuture类实现更高效的异步处理，

总结：

1.流不仅仅是将外部迭代变为内部迭代，更是一种编程思想的转变，结合函数抽象，行为参数化，将函数作为参数，提升为与值一样的地位，威力巨大，这就是生产力。

2.流计算能大大简化编程，使用声明式语法，配合Lambda，写起代码根本停不下来！

3.后期会在看看其他Storm/Spark/Flink流式计算框架，发现点新鲜货。

全文结束！

---

推荐阅读：

• Dubbo学习系列之十六（ELK海量日志分析）
• Linux下Redis集群
• Dubbo学习系列之十五（Seata分布式事务方案TCC模式）
• Dubbo学习系列之十四（Seata分布式事务方案AT模式）
• Dubbo学习系列之十三（Mycat数据库代理）