Stream流处理快速上手最佳实践 | 京东物流技术团队

京东云开发者
• 阅读 414

一 引言

JAVA1.8得益于Lambda所带来的函数式编程,引入了一个全新的Stream流概念Stream流式思想类似于工厂车间的“生产流水线”,Stream流不是一种数据结构,不保存数据,而是对数据进行加工处理。Stream可以看作是流水线上的一个工序。在流水线上,通过多个工序让一个原材料加工成一个商品。

二 常用方法介绍

2.1 获取Stream流

所有的 Collection 集合都可以通过 stream 默认方法获取流;

java.util.Collection 接口中加入了default方法 stream 用来获取流,所以其所有实现类均可获取流。

ArrayList<XyBug> xyBugList = new ArrayList();
Stream<XyBug> stream = xyBugList.stream();


Stream 接口的静态方法 of 可以获取数组对应的流。

//String
Stream<String> stream = Stream.of("aa", "bb", "cc");
//数组
String[] arr = {"aa", "bb", "cc"};
Stream<String> stream7 = Stream.of(arr);
Integer[] arr2 = {11, 22, 33};
Stream<Integer> stream8 = Stream.of(arr2);
//对象
XyBug xyBug1 = new XyBug();
XyBug xyBug2 = new XyBug();
XyBug xyBug3 = new XyBug();
Stream<XyBug> bugStream = Stream.of(xyBug1, xyBug2, xyBug3);



2.2 Stream 数据处理常用方法

forEach方法

该方法接收一个 Consumer 接口函数,会将每一个流元素交给该函数进行处理

List<String> list = new ArrayList<>();
Collections.addAll(list, "str1", "str2", "str3", "str4", "str5", "str6");
list.stream().forEach((String s) -> {
  System.out.println(s);
  });
//简写
list.stream().forEach(s -> System.out.println(s));


s代表list中的每一个元素,流式处理依次遍历每个元素

->后的代码为每个元素处理逻辑

count方法

count 方法来统计其中的元素个数,返回值为long类型

long count = list.stream().count();


distinct方法

对流中的数据进行去重操作,普通类型可直接去重

Stream流处理快速上手最佳实践 | 京东物流技术团队

//将22、33重复数据去除
Stream.of(22, 33, 22, 11, 33).distinct().collect(Collectors.toList());


自定义类型是根据对象的hashCode和equals来去除重复元素的

XyBug实体类中加@Data注解,hashCode和equals会别重写,在使用distinct方法时判断去重

ArrayList bugList = JSON.parseObject(bugs, ArrayList.class);
ArrayList<XyBug> xyBugList = new ArrayList();
List collect = (List) bugList.stream().distinct().collect(Collectors.toList());


通过distinct()方法去重,去重后的数据通过collect(Collectors.toList())组成新6的list

limit方法

方法可以对流进行截取,只取用前n个,参数是一个long型,如果集合当前长度大于参数则进行截取。否则不进行操作

Stream流处理快速上手最佳实践 | 京东物流技术团队

List<String> list = new ArrayList<>();
Collections.addAll(list, "1", "2", "3", "4", "5", "6");
List<String> collect = list.stream().limit(3).collect(Collectors.toList());


将前3个String对象截取,组成新的list

skip方法

如果希望跳过前几个元素,可以使用 skip 方法获取一个截取之后的新流,如果流的当前长度大于n,则跳过前n个;否则将会得到一个长度为0的空流

List<String> list = new ArrayList<>();
Collections.addAll(list, "1", "2", "3", "4", "5", "6");
List<String> collect = list.stream().skip(3).collect(Collectors.toList());


跳过前3个String对象,后三个组成新的list

filter方法

filter用于过滤数据,返回符合过滤条件的数据,可以通过 filter 方法将一个流转换成另一个子集流,该接口接收一个 Predicate 函数式接口参数(可以是一个Lambda或方法引用)作为筛选条件

List<String> list = new ArrayList<>();
Collections.addAll(list, "1", "22", "3", "4", "55", "6");
//filter方法中写入筛选条件,将过滤后的数据组成新的list
list.stream().filter(s -> s.length() == 2).collect(Collectors.toList());


通过该条语句s -> s.length() == 2,筛选出22、55

map方法

将流中的元素映射到另一个流中,可以将当前流中的T类型数据转换为另一种R类型的流

List<PersonCrDto> laputaCrDtos = queryListLaputaByBeginEndTime(begin, end);
//将list中的PersonCrDto对象的userName属性取到,收集成set集合
laputaCrDtos.stream().map(PersonCrDto::getUserName).collect(Collectors.toSet())


将list中的每个对象的userName数据拿到,组成Set集合

stream分组

List<XyBug> list = new ArrayList<>();
Map<String, List<XyBug>> collect = list.stream().collect(Collectors.groupingBy(XyBug::getBugType));


根据bug类型进行分组,分组后会组成map,key是组名,value是组下的数据

stream排序

sort(),默认正序排列,加入reversed()方法后倒叙排列

List<XyBug> list = new ArrayList<>();
//根据createTime正序排列
List<XyBug> collect = list.stream().sorted(Comparator.comparing(XyBug::getCreateTime)).collect(Collectors.toList());
//根据createTime倒叙排列
List<XyBug> collect = list.stream().sorted(Comparator.comparing(XyBug::getCreateTime).reversed()).collect(Collectors.toList());



collect方法

将处理后数据收集为list,collect(Collectors.toList())

将处理后数据收集为set,collect(Collectors.toSet())

根据某个字段值将数据分组map,collect(Collectors.groupingBy(o -> o.value())))

三 实践举例

需求:将bug数据通过orgTierName分组,存储到map中

未使用Stream,需要使用for循环并且进行各种判断,代码行数较多

HashMap<String, List<XyBug>> map = new HashMap<>();
for (XyBug one : bugList){
    if(one.getOrgTierName() != null){
        if(map.get(one.getOrgTierName()) == null){
            List<XyBug> list = new ArrayList();
            list.add(one);
            map.put(one.getOrgTierName(),list);
        }else {
            map.get(one.getOrgTierName()).add(one);
        }
    }
}


使用Stream,一行代码搞定,直观并高效

collectDeptBugMap = bugList.stream().filter(o -> o.getOrgTierName() != null).collect(Collectors.groupingBy(o -> o.getOrgTierName()));


四 总结

Stream是对集合(Collection)对象功能的增强,能对集合对象进行各种非常便利、高效的聚合操作,或者大批量数据操作,提高编程效率、简洁性和程序可读性。本文通过简单举例,希望帮助读者快速上手使用流处理,Stream流处理功能非常强全,更多方法请参考API文档。

作者:京东物流 杨靖平

来源:京东云开发者社区 自猿其说Tech 转载请注明来源

点赞
收藏
评论区
推荐文章
执键写春秋 执键写春秋
3年前
Stream流式处理&&Lambda表达式
高度抽象的集合数据处理1.使用Stream流对List进行去重、去偶数,然后查找最大值importjava.util.Arrays;importjava.util.Comparator;publicclassStream1publicstaticvoidmain(Stringargs)intmaxA
Wesley13 Wesley13
3年前
Java Stream
1Stream简介Stream是数据渠道,用于操作数据源(集合,数组等)所生成得元素序列。而集合讲得是数据,流讲得是计算。注意:Stream自己不会存储元素。Stream不会改变源对象。相反,它会返回一个持有结果得新StreamStream操作时延迟执行得,这意味着它们会等到需要结果时才执
Wesley13 Wesley13
3年前
JDK1.8 之Stream API总结
Stream是Java8新增加的类,用来补充集合类。Stream代表数据流,流中的数据元素的数量可能是有限的,也可能是无限的。Stream和其它集合类的区别在于:其它集合类主要关注与有限数量的数据的访问和有效管理(增删改),而Stream并没有提供访问和管理元素的方式,而是通过声明数据源的方式,利用可计算的操作在数据源上执行,当然
Wesley13 Wesley13
3年前
Java8系列之Stream总结
流的简介  官方解释,Stream是Java8的一大亮点,它与java.io包里的InputStream和OutputStream是完全不同的概念。它也不同于StAX对XML的解析的Stream,也不是AmazonKinesis对大数据实时处理的Stream。它是对集合对象功能的增强,她专注于对集合对象进行各种非常便利、高效的聚合操作(ag
Wesley13 Wesley13
3年前
Java学习:Stream流式思想
Stream流Java8API添加了一种新的机制——Stream(流)。Stream和IO流不是一回事。流式思想:像生产流水线一样,一个操作接一个操作。使用Stream流的步骤:数据源→转换成流→操作1→操作2→……数据源(source):可以是集合、数组等。St
Wesley13 Wesley13
3年前
Java 8新特性之Stream 概念
Java8中有两大最为重要的改变。第一个是Lambda表达式;另外一个则是StreamAPI(java.util.stream.\)。Stream是Java8中处理集合的关键抽象概念,它可以指定你希望对集合进行的操作,可以执行非常复杂的查找、过滤和映射数据等操作。使用StreamAPI对集合数据进行操作,就类似于使用SQL执行
Wesley13 Wesley13
3年前
Java8 新特性之集合操作Stream
Java8新特性之集合操作StreamStream简介Java8引入了全新的StreamAPI。这里的Stream和I/O流不同,它更像具有Iterable的集合类,但行为和集合类又有所不同。stream是对集合对象功能的增强,它专注于对集合对象进行各种非常便利、高效的聚合操作,或者大批量数据操作。
Easter79 Easter79
3年前
Trident API(翻译)
TridentAPIOverviewTrident的核心数据模型是“流”(Stream),进行数据处理的时候,将数据作为一系列的batch(批)来进行。流被分割成多个partition分布在集群中的不同节点上来运行,而且对流的操作也是在流的各个partition上并行运行的。Trident中有五类操作
Wesley13 Wesley13
3年前
Java 8 Stream API学习总结
Java8API添加了一个新的抽象称为流Stream,可以让你以一种声明的方式处理数据。StreamAPI可以极大提高Java程序员的生产力,让程序员写出高效率、干净、简洁的代码。这种风格将要处理的元素集合看作一种流,流在管道中传输,并且可以在管道的节点上进行处理,比如筛选,排序,聚合等。元素流在管道中经过中间操作(intermediateo
京东云开发者 京东云开发者
10个月前
使用Flink完成流数据统计 | 京东云技术团队
Flink程序构建的基本单元是stream和transformation(DataSet实质上也是stream)。stream是一个中间结果数据,transformation对数据的加工和操作,该操作以一个或多个stream为输入,计算输出一个或多个stream为结果,最后可以sink来存储数据。