(92) 函数式数据处理 (上) / 计算机程序的思维逻辑

上节我们介绍了Lambda表达式和函数式接口,本节探讨它们的应用,函数式数据处理,针对常见的集合数据处理,Java 8引入了一套新的类库,位于包java.util.stream下,称之为Stream API,这套API操作数据的思路,不同于我们在38节55节介绍的容器类API,它们是函数式的,非常简洁、灵活、易读,具体有什么不同呢?由于内容较多,我们分为两节来介绍,本节先介绍一些基本的API,下节讨论一些高级功能。

基本概念

接口Stream类似于一个迭代器,但提供了更为丰富的操作,Stream API的主要操作就定义在该接口中。 Java 8给Collection接口增加了两个默认方法,它们可以返回一个Stream,如下所示:

default Stream<E> stream() { return StreamSupport.stream(spliterator(), false); } default Stream<E> parallelStream() { return StreamSupport.stream(spliterator(), true); }

stream()返回的是一个顺序流,parallelStream()返回的是一个并发流。顺序流就是由一个线程执行操作。而并发流背后可能有多个线程并发执行,与之前介绍的并发技术不同,使用并发流不需要显式管理线程,使用方法与顺序流是一样的。

下面,我们主要针对顺序流,学习Stream接口,包括其用法和基本原理,随后我们再介绍下并发流。先来看一些简单的示例。 基本示例 上节演示时使用了学生类Student和学生列表List<Student> lists,本节继续使用它们。 基本过滤 返回学生列表中90分以上的,传统上的代码一般是这样的:

List<Student> above90List = new ArrayList<>(); for (Student t : students) { if (t.getScore() > 90) { above90List.add(t); } }

使用Stream API,代码可以这样:

List<Student> above90List = students.stream() .filter(t->t.getScore()>90) .collect(Collectors.toList());

先通过stream()得到一个Stream对象,然后调用Stream上的方法,filter()过滤得到90分以上的,它的返回值依然是一个Stream,为了转换为List,调用了collect方法并传递了一个Collectors.toList(),表示将结果收集到一个List中。 代码更为简洁易读了,这种数据处理方式被称为函数式数据处理,与传统代码相比,它的特点是:

  • 没有显式的循环迭代,循环过程被Stream的方法隐藏了
  • 提供了声明式的处理函数,比如filter,它封装了数据过滤的功能,而传统代码是命令式的,需要一步步的操作指令
  • 流畅式接口,方法调用链接在一起,清晰易读

基本转换 根据学生列表返回名称列表,传统上的代码一般是这样:

List<String> nameList = new ArrayList<>(students.size()); for (Student t : students) { nameList.add(t.getName()); }

使用Stream API,代码可以这样:

List<String> nameList = students.stream() .map(Student::getName) .collect(Collectors.toList());

这里使用了Stream的map函数,它的参数是一个Function函数式接口,这里传递了方法引用。 基本的过滤和转换组合 返回90分以上的学生名称列表,传统上的代码一般是这样:

List<String> nameList = new ArrayList<>(); for (Student t : students) { if (t.getScore() > 90) { nameList.add(t.getName()); } }

使用函数式数据处理的思路,可以将这个问题分解为由两个基本函数实现:

  1. 过滤:得到90分以上的学生列表
  2. 转换:将学生列表转换为名称列表

使用Stream API,可以将基本函数filter()和map()结合起来,代码可以这样:

List<String> above90Names = students.stream() .filter(t->t.getScore()>90) .map(Student::getName) .collect(Collectors.toList());

这种组合利用基本函数、声明式实现集合数据处理功能的编程风格,就是函数式数据处理。 代码更为直观易读了,但你可能会担心它的性能有问题。filter()和map()都需要对流中的每个元素操作一次,一起使用会不会就需要遍历两次呢?答案是否定的,只需要一次。实际上,调用filter()和map()都不会执行任何实际的操作,它们只是在构建操作的流水线,调用collect才会触发实际的遍历执行,在一次遍历中完成过滤、转换以及收集结果的任务。 像filter和map这种不实际触发执行、用于构建流水线、返回Stream的操作被称为中间操作(intermediate operation),而像collect这种触发实际执行、返回具体结果的操作被称为终端操作(terminal operation)。Stream API中还有更多的中间和终端操作,下面我们具体来看下。 中间操作 除了filter和map,Stream API的中间操作还有distinct, sorted, skip, limit, peek, mapToLong, mapToInt, mapToDouble, flatMap等,我们逐个来看下。 distinct distinct返回一个新的Stream,过滤重复的元素,只留下唯一的元素,是否重复是根据equals方法来比较的,distinct可以与其他函数如filter, map结合使用。

比如,返回字符串列表中长度小于3的字符串、转换为小写、只保留唯一的,代码可以为:

List<String> list = Arrays.asList(new String[]{"abc","def","hello","Abc"}); List<String> retList = list.stream() .filter(s->s.length()<=3) .map(String::toLowerCase) .distinct() .collect(Collectors.toList()); System.out.println(retList);

输出为:

[abc, def]

虽然都是中间操作,但distinct与filter和map是不同的,filter和map都是无状态的,对于流中的每一个元素,它的处理都是独立的,处理后即交给流水线中的下一个操作,但distinct不同,它是有状态的,在处理过程中,它需要在内部记录之前出现过的元素,如果已经出现过,即重复元素,它就会过滤掉,不传递给流水线中的下一个操作。 对于顺序流,内部实现时,distinct操作会使用HashSet记录出现过的元素,如果流是有顺序的,需要保留顺序,会使用LinkedHashSetsorted 有两个sorted方法:

Stream<T> sorted() Stream<T> sorted(Comparator<? super T> comparator)

它们都对流中的元素排序,都返回一个排序后的Stream,第一个方法假定元素实现了Comparable接口,第二个方法接受一个自定义的Comparator。

比如,过滤得到90分以上的学生,然后按分数从高到低排序,分数一样的,按名称排序,代码可以为:

List<Student> list = students.stream() .filter(t->t.getScore()>90) .sorted(Comparator.comparing(Student::getScore) .reversed() .thenComparing(Student::getName)) .collect(Collectors.toList());

这里,使用了Comparator的comparing, reversed和thenComparing构建了Comparator。 与distinct一样,sorted也是一个有状态的中间操作,在处理过程中,需要在内部记录出现过的元素,与distinct不同的是,每碰到流中的一个元素,distinct都能立即做出处理,要么过滤,要么马上传递给下一个操作,但sorted不能,它需要先排序,为了排序,它需要先在内部数组中保存碰到的每一个元素,到流结尾时,再对数组排序,然后再将排序后的元素逐个传递给流水线中的下一个操作。 skip/limit 它们的定义为:

Stream<T> skip(long n) Stream<T> limit(long maxSize)

skip跳过流中的n个元素,如果流中元素不足n个,返回一个空流,limit限制流的长度为maxSize。

比如,将学生列表按照分数排序,返回第3名到第5名,代码可以为:

List<Student> list = students.stream() .sorted(Comparator.comparing( Student::getScore).reversed()) .skip(2) .limit(3) .collect(Collectors.toList());

skip和limit都是有状态的中间操作。对前n个元素,skip的操作就是过滤,对后面的元素,skip就是传递给流水线中的下一个操作。limit的一个特点是,它不需要处理流中的所有元素,只要处理的元素个数达到maxSize,后面的元素就不需要处理了,这种可以提前结束的操作被称为短路操作。 peek peek的定义为:

Stream<T> peek(Consumer<? super T> action)

它返回的流与之前的流是一样的,没有变化,但它提供了一个Consumer,会将流中的每一个元素传给该Consumer。这个方法的主要目的是支持调试,可以使用该方法观察在流水线中流转的元素,比如:

List<String> above90Names = students.stream() .filter(t->t.getScore()>90) .peek(System.out::println) .map(Student::getName) .collect(Collectors.toList());

mapToLong/mapToInt/mapToDouble map函数接受的参数是一个Function<T, R>,为避免装箱/拆箱,提高性能,Stream还有如下返回基本类型特定流的方法:

DoubleStream mapToDouble(ToDoubleFunction<? super T> mapper) IntStream mapToInt(ToIntFunction<? super T> mapper) LongStream mapToLong(ToLongFunction<? super T> mapper)

DoubleStream/IntStream/LongStream是基本类型特定的流,有一些专门的更为高效的方法。比如,求学生列表的分数总和,代码可以为:

double sum = students.stream() .mapToDouble(Student::getScore) .sum();

flatMap flatMap的定义为:

<R> Stream<R> flatMap(Function<? super T, ? extends Stream<? extends R>> mapper)

它接受一个函数mapper,对流中的每一个元素,mapper会将该元素转换为一个流Stream,然后把新生成流的每一个元素传递给下一个操作。比如:

List<String> lines = Arrays.asList(new String[]{ "hello abc", "老马 编程" }); List<String> words = lines.stream() .flatMap(line -> Arrays.stream(line.split("\\s+"))) .collect(Collectors.toList()); System.out.println(words);

这里的mapper将一行字符串按空白符分隔为了一个单词流,Arrays.stream可以将一个数组转换为一个流,输出为:

[hello, abc, 老马, 编程]

可以看出,实际上,flatMap完成了一个1到n的映射。 针对基本类型,flatMap还有如下类似方法:

DoubleStream flatMapToDouble(Function<? super T, ? extends DoubleStream> mapper) IntStream flatMapToInt(Function<? super T, ? extends IntStream> mapper) LongStream flatMapToLong(Function<? super T, ? extends LongStream> mapper)

终端操作 中间操作不触发实际的执行,返回值是Stream,而终端操作触发执行,返回一个具体的值,除了collect,Stream API的终端操作还有max, min, count, allMatch, anyMatch, noneMatch, findFirst, findAny, forEach, toArray, reduce等,我们逐个来看下。 max/min max/min的定义为:

Optional<T> max(Comparator<? super T> comparator) Optional<T> min(Comparator<? super T> comparator)

它们返回流中的最大值/最小值,值的注意的是,它的返回值类型是Optional<T>,而不是T。 java.util.Optional是Java 8引入的一个新类,它是一个泛型容器类,内部只有一个类型为T的单一变量value,可能为null,也可能不为null。Optional有什么用呢?它用于准确地传递程序的语义,它清楚地表明,其代表的值可能为null,程序员应该进行适当的处理。

Optional定义了一些方法,比如:

// value不为null时返回true public boolean isPresent() // 返回实际的值,如果为null,抛出异常NoSuchElementException public T get() // 如果value不为null,返回value,否则返回other public T orElse(T other) // 构建一个空的Optional,value为null public static<T> Optional<T> empty() // 构建一个非空的Optional, 参数value不能为null public static <T> Optional<T> of(T value) // 构建一个Optional,参数value可以为null,也可以不为null public static <T> Optional<T> ofNullable(T value)

在max/min的例子中,通过声明返回值为Optional,我们就知道,具体的返回值不一定存在,这发生在流中不含任何元素的情况下。 看个简单的例子,返回分数最高的学生,代码可以为:

Student student = students.stream() .max(Comparator.comparing(Student::getScore).reversed()) .get();

这里,假定students不为空。

count count很简单,就是返回流中元素的个数。比如,统计大于90分的学生个数,代码可以为:

long above90Count = students.stream() .filter(t->t.getScore()>90) .count();

allMatch/anyMatch/noneMatch 这几个函数都接受一个谓词Predicate,返回一个boolean值,用于判定流中的元素是否满足一定的条件,它们的区别是:

  • allMatch: 只有在流中所有元素都满足条件的情况下才返回true
  • anyMatch: 只要流中有一个元素满足条件就返回true
  • noneMatch: 只有流中所有元素都不满足条件才返回true

如果流为空,这几个函数的返回值都是true。 比如,判断是不是所有学生都及格了(不小于60分),代码可以为:

boolean allPass = students.stream() .allMatch(t->t.getScore()>=60);

这几个操作都是短路操作,都不一定需要处理所有元素就能得出结果,比如,对于allMatch,只要有一个元素不满足条件,就能返回false。 findFirst/findAny 它们的定义为:

Optional<T> findFirst() Optional<T> findAny()

它们的返回类型都是Optional,如果流为空,返回Optional.empty()。findFirst返回第一个元素,而findAny返回任一元素,它们都是短路操作。 随便找一个不及格的学生,代码可以为:

Optional<Student> student = students.stream() .filter(t->t.getScore()<60) .findAny(); if(student.isPresent()){ // 不及格的学生.... }

forEach 有两个foreach方法:

void forEach(Consumer<? super T> action) void forEachOrdered(Consumer<? super T> action)

它们都接受一个Consumer,对流中的每一个元素,传递元素给Consumer,区别在于,在并发流中,forEach不保证处理的顺序,而forEachOrdered会保证按照流中元素的出现顺序进行处理。 比如,逐行打印大于90分的学生,代码可以为:

students.stream() .filter(t->t.getScore()>90) .forEach(System.out::println);

toArray toArray将流转换为数组,有两个方法:

Object[] toArray() <A> A[] toArray(IntFunction<A[]> generator)

不带参数的toArray返回的数组类型为Object[],这经常不是期望的结果,如果希望得到正确类型的数组,需要传递一个类型为IntFunction的generator,IntFunction的定义为:

public interface IntFunction<R> { R apply(int value); }

generator接受的参数是流的元素个数,它应该返回对应大小的正确类型的数组。 比如,获取90分以上的学生数组,代码可以为:

Student[] above90Arr = students.stream() .filter(t->t.getScore()>90) .toArray(Student[]::new);

Student[]::new就是一个类型为IntFunction<Student[]>的generator。 reduce reduce代表归约或者叫折叠,它是max/min/count的更为通用的函数,将流中的元素归约为一个值,有三个reduce函数:

Optional<T> reduce(BinaryOperator<T> accumulator); T reduce(T identity, BinaryOperator<T> accumulator); <U> U reduce(U identity, BiFunction<U, ? super T, U> accumulator, BinaryOperator<U> combiner);

第一个基本等同于调用:

boolean foundAny = false; T result = null; for (T element : this stream) { if (!foundAny) { foundAny = true; result = element; } else result = accumulator.apply(result, element); } return foundAny ? Optional.of(result) : Optional.empty();

比如,使用reduce求分数最高的学生,代码可以为:

Student topStudent = students.stream().reduce((accu, t) -> { if (accu.getScore() >= t.getScore()) { return accu; } else { return t; } }).get();

第二个reduce函数多了一个identity参数,表示初始值,它基本等同于调用:

T result = identity; for (T element : this stream) result = accumulator.apply(result, element) return result;

第一个和第二个reduce的返回类型只能是流中元素的类型,而第三个更为通用,它的归约类型可以自定义,另外,它多了一个combiner参数,combiner用在并发流中,用于合并子线程的结果,对于顺序流,它基本等同于调用:

U result = identity; for (T element : this stream) result = accumulator.apply(result, element) return result;

注意与第二个reduce函数相区分,它的结果类型不是T,而是U。比如,使用reduce函数计算学生分数的和,代码可以为:

double sumScore = students.stream().reduce(0d, (sum, t) -> sum += t.getScore(), (sum1, sum2) -> sum1 += sum2 );

以上,可以看出,reduce虽然更为通用,但比较费解,难以使用,一般情况,应该优先使用其他函数。 collect函数比reduce更为通用、强大和易用,关于它,我们下节再详细介绍。 构建流 前面我们提到,可以通过Collection接口的stream/parallelStream获取流,还有一些其他的方式可以获取流。 Arrays有一些stream方法,可以将数组或子数组转换为流,比如:

public static IntStream stream(int[] array) public static DoubleStream stream(double[] array, int startInclusive, int endExclusive) public static <T> Stream<T> stream(T[] array)

比如,输出当前目录下所有普通文件的名字,代码可以为:

File[] files = new File(".").listFiles(); Arrays.stream(files) .filter(File::isFile) .map(File::getName) .forEach(System.out::println);

Stream也有一些静态方法,可以构建流:

//返回一个空流 public static<T> Stream<T> empty() //返回只包含一个元素t的流 public static<T> Stream<T> of(T t) //返回包含多个元素values的流 public static<T> Stream<T> of(T... values) //通过Supplier生成流,流的元素个数是无限的 public static<T> Stream<T> generate(Supplier<T> s) //同样生成无限流,第一个元素为seed,第二个为f(seed),第三个为f(f(seed)),依次类推 public static<T> Stream<T> iterate(final T seed, final UnaryOperator<T> f)

比如,输出10个随机数,代码可以为:

Stream.generate(()->Math.random()) .limit(10) .forEach(System.out::println);

输出100个递增的奇数,代码可以为:

Stream.iterate(1, t->t+2) .limit(100) .forEach(System.out::println);

并发流 前面我们主要使用的是Collection的stream()方法,换做parallelStream()方法,就会使用并发流,接口方法都是通用的。但并发流内部会使用多线程,线程个数一般与系统的CPU核数一样,以充分利用CPU的计算能力。

进一步来说,并发流内部会使用Java 7引入的fork/join框架,简单来说,处理由fork和join两个阶段组成,fork就是将要处理的数据拆分为小块,多线程按小块进行并发计算,join就是将小块的计算结果进行合并,具体我们就不探讨了。使用并发流,不需要任何线程管理的代码,就能实现并发。 函数式数据处理思维 看的出来,使用Stream API处理数据集合,与直接使用容器类API处理数据的思路是完全不一样的。 流定义了很多数据处理的基本函数,对于一个具体的数据处理问题,解决的主要思路就是组合利用这些基本函数,实现期望的功能,这种思路就是函数式数据处理思维,相比直接利用容器类API的命令式思维,思考的层次更高。

Stream API的这种思路也不是新发明,它与数据库查询语言SQL是很像的,都是声明式地操作集合数据,很多函数都能在SQL中找到对应,比如filter对应SQL的where,sorted对应order by等。SQL一般都支持分组(group by)功能,Stream API也支持,但关于分组,我们下节再介绍。 Stream API也与各种基于Unix系统的管道命令类似,熟悉Unix系统的都知道,Unix有很多命令,大部分命令只是专注于完成一件事情,但可以通过管道的方式将多个命令链接起来,完成一些复杂的功能,比如:

cat nginx_access.log | awk '{print $1}' | sort | uniq -c | sort -rnk 1 | head -n 20

以上命令可以分析nginx访问日志,统计出访问次数最多的前20个IP地址及其访问次数。具体来说,cat命令输出nginx访问日志到流,一行为一个元素,awk输出行的第一列,这里为IP地址,sort按IP进行排序,"uniq -c"按IP统计计数,"sort -rnk 1"按计数从高到低排序,"head -n 20"输出前20行。 小结 本节初步介绍了Java 8引入的函数式数据处理类库,Stream API,它类似于Unix的管道命令,也类似于数据库查询语言SQL,通过组合利用基本函数,可以在更高的层次上思考问题,以声明式的方式简洁地实现期望的功能。 对于collect方法,本节只是演示了最基本的应用,它还有很多高级功能,比如实现类似SQL的group by功能,具体怎么实现?实现的原理是什么呢?

(与其他章节一样,本节所有代码位于 https://github.com/swiftma/program-logic,位于包shuo.laoma.java8.c92下)

原文发布于微信公众号 - 老马说编程(laoma_shuo)

原文发表时间:2017-06-27

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

发表于

我来说两句

0 条评论
登录 后参与评论

相关文章

来自专栏屈定‘s Blog

Java--Enum的思考

枚举类是Java5引进的特性,其目的是替换int枚举模式或者String枚举模式,使得语义更加清晰,另外也解决了行为和枚举绑定的问题.

1394
来自专栏Pythonista

封装与扩展性

封装在于明确区分内外,使得类实现者可以修改封装内的东西而不影响外部调用者的代码;而外部使用用者只知道一个接口(函数),只要接口(函数)名、参数不变,使用者的代码...

933
来自专栏用户2442861的专栏

json格式

  1. “名称/值”对的集合(A collection of name/value pairs)。不同的语言中,它被理解为对象(object),记录(reco...

2442
来自专栏上善若水

s002android逆向安全初级篇之android smali语法总结

smali中有两类数据类型:基本类型和引用类型。 引用类型是指数组和对象,其他都是基础类型。

2224
来自专栏Web 开发

JavaScript的对象引用

在一个函数体内,var变量声明的变量,其作用域只在该函数体内,对于函数体外而言,是不可见的(废话)。

810
来自专栏编程

宝宝都能学会的python编程教程4:关系运算符与循环

关系运算符 if 语句 实际应用中的程序,大多不是一撮而就的,而是根据条件不同存在很多分支。 最基本的条件分支结构是if...else...语句即如果。。。否则...

2139
来自专栏菜鸟计划

angularjs filter详解

过滤器(filter)正如其名,作用就是接收一个输入,通过某个规则进行处理,然后返回处理后的结果。 主要用在数据的格式化上,例如获取一个数组中的子集,对数组中的...

3728
来自专栏余林丰

Java中net.sf.json包关于JSON与对象互转的坑

  在Web开发过程中离不开数据的交互,这就需要规定交互数据的相关格式,以便数据在客户端与服务器之间进行传递。数据的格式通常有2种:1、xml;2、JSON。通...

3595
来自专栏玄魂工作室

如何学python 第十一课 元组与字典

第十一课 元组与字典 欢迎回来。上一期的如何学python里,我们讨论了函数。我们今天将要学习的是两种类似于列表(list)类型的数据类型。我们先介绍’元组’...

3144
来自专栏老司机的技术博客

人人都能学会的python编程教程4:关系运算符与循环

在python当中,if condition1:(注意最后的冒号:)称为“语句头”。冒号:之后另起一行缩进的是“语句体”,语句体的行数不限,但至少有一行,否则需...

3429

扫码关注云+社区

领取腾讯云代金券