zoukankan      html  css  js  c++  java
  • 流式计算(一)-Java8Stream

    大约各位看官君多少也听说了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流式计算框架,发现点新鲜货。

    全文结束!


    推荐阅读:

  • 相关阅读:
    计算广告学学习1
    scala学习手记20
    scala学习手记19
    scala学习手记18
    scala学习手记17
    SAM4E单片机之旅——12、USART
    SAM4E单片机之旅——11、UART之PDC收发
    SAM4E单片机之旅——10、UART与MCK之PLL
    SAM4E单片机之旅——9、UART与MCK之MAINCK
    SAM4E单片机之旅——8、UART初步
  • 原文地址:https://www.cnblogs.com/xxbiao/p/12048946.html
Copyright © 2011-2022 走看看