码迷,mamicode.com
首页 > 编程语言 > 详细

流式计算(一)-Java8Stream

时间:2019-12-16 15:00:46      阅读:829      评论:0      收藏:0      [点我收藏+]

标签:编程思想   tab   先来   one   应用   find   分布式   tst   新手   

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

全文结束!

技术图片


推荐阅读:

流式计算(一)-Java8Stream

标签:编程思想   tab   先来   one   应用   find   分布式   tst   新手   

原文地址:https://www.cnblogs.com/xxbiao/p/12048946.html

(0)
(0)
   
举报
评论 一句话评论(0
登录后才能评论!
© 2014 mamicode.com 版权所有  联系我们:gaon5@hotmail.com
迷上了代码!