Java8中的Stream详解

为什么需要 Stream

Stream 作为 Java 8 的一大亮点,它与 java.io 包里的 InputStream 和 OutputStream 是完全不同的概念。它也不同于 StAX 对 XML 解析的 Stream,也不是 Amazon Kinesis 对大数据实时处理的 Stream。Java 8 中的 Stream 是对集合(Collection)对象功能的增强,它专注于对集合对象进行各种非常便利、高效的聚合操作(aggregate operation),或者大批量数据操作 (bulk data operation)。Stream API 借助于同样新出现的 Lambda 表达式,极大的提高编程效率和程序可读性。同时它提供串行和并行两种模式进行汇聚操作,并发模式能够充分利用多核处理器的优势,使用 fork/join 并行方式来拆分任务和加速处理过程。通常编写并行代码很难而且容易出错, 但使用 Stream API 无需编写一行多线程的代码,就可以很方便地写出高性能的并发程序。所以说,Java 8 中首次出现的 java.util.stream 是一个函数式语言+多核时代综合影响的产物。


什么是流

Stream 不是集合元素,它不是数据结构并不保存数据,它是有关算法和计算的,它更像一个高级版本的 Iterator。

Stream 就如同一个迭代器(Iterator),单向,不可往复,数据只能遍历一次,遍历过一次后即用尽了,就好比流水从面前流过,一去不复返。

而和迭代器又不同的是,Stream 可以并行化操作,Stream 的并行操作依赖于 Java7 中引入的 Fork/Join 框架(JSR166y)来拆分任务和加速处理过程。Java 的并行 API 演变历程基本如下:

  1. 1.0-1.4 中的 java.lang.Thread
  2. 5.0 中的 java.util.concurrent
  3. 6.0 中的 Phasers 等
  4. 7.0 中的 Fork/Join 框架
  5. 8.0 中的 Lambda

Stream 的另外一大特点是,数据源本身可以是无限的。


流的构成

当我们使用一个流的时候,通常包括三个基本步骤:

获取一个数据源(source)→ 数据转换→执行操作获取想要的结果

有多种方式生成 Stream Source:

  • 从 Collection 和数组

    • Collection.stream()
    • Collection.parallelStream()
    • Arrays.stream(T array) or Stream.of()

    从 BufferedReader

    • java.io.BufferedReader.lines()
  • 静态工厂

  • java.util.stream.IntStream.range()

  • java.nio.file.Files.walk()

  • 自己构建

    • java.util.Spliterator

    其它

    • Random.ints()
    • BitSet.stream()
    • Pattern.splitAsStream(java.lang.CharSequence)
    • JarFile.stream()

流的操作类型分为两种:

  • **Intermediate(中间)**:一个流可以后面跟随零个或多个 intermediate 操作。其目的主要是打开流,做出某种程度的数据映射/过滤,然后返回一个新的流,交给下一个操作使用。这类操作都是惰性化的(lazy),就是说,仅仅调用到这类方法,并没有真正开始流的遍历。
  • **Terminal(终点)**:一个流只能有一个 terminal 操作,当这个操作执行后,流就被使用“光”了,无法再被操作。所以这必定是流的最后一个操作。

在对于一个 Stream 进行多次转换操作 (Intermediate 操作),每次都对 Stream 的每个元素进行转换,而且是执行多次,这样时间复杂度就是 N(转换次数)个 for 循环里把所有操作都做掉的总和吗?其实不是这样的,转换操作都是 lazy 的,多个转换操作只会在 Terminal 操作的时候融合起来,一次循环完成。我们可以这样简单的理解,Stream 里有个操作函数的集合,每次转换操作就是把转换函数放入这个集合中,在 Terminal 操作的时候循环 Stream 对应的集合,然后对每个元素执行所有的函数。

还有一种操作被称为 short-circuiting。用以指:

  • 对于一个 intermediate 操作,如果它接受的是一个无限大(infinite/unbounded)的 Stream,但返回一个有限的新 Stream。
  • 对于一个 terminal 操作,如果它接受的是一个无限大的 Stream,但能在有限的时间计算出结果。

当操作一个无限大的 Stream,而又希望在有限时间内完成操作,则在管道内拥有一个 short-circuiting 操作是必要非充分条件。


流的构造与转换

简单说,对 Stream 的使用就是实现一个 filter-map-reduce 过程,产生一个最终结果,或者导致一个副作用(side effect)。

下面提供最常见的几种构造 Stream 的样例。

构造流的几种常见方法
1
2
3
4
5
6
7
8
9
// 1. Individual values
Stream stream = Stream.of("a", "b", "c");
// 2. Arrays
String [] strArray = new String[] {"a", "b", "c"};
stream = Stream.of(strArray);
stream = Arrays.stream(strArray);
// 3. Collections
List<``String``> list = Arrays.asList(strArray);
stream = list.stream();

需要注意的是,对于基本数值型,目前有三种对应的包装类型 Stream:

IntStream、LongStream、DoubleStream。当然我们也可以用 Stream、Stream >、Stream,但是 boxing 和 unboxing 会很耗时,所以特别为这三种基本数值型提供了对应的 Stream。

Java 8 中还没有提供其它数值型 Stream,因为这将导致扩增的内容较多。而常规的数值型聚合运算可以通过上面三种 Stream 进行。

数值流的构造
1
2
3
IntStream.of(new int[]{1, 2, 3}).forEach(System.out::println);
IntStream.range(1, 3).forEach(System.out::println);
IntStream.rangeClosed(1, 3).forEach(System.out::println);
流转换为其它数据结构
1
2
3
4
5
6
7
8
9
// 1. Array
String[] strArray1 = stream.toArray(String[]::new);
// 2. Collection
List<``String``> list1 = stream.collect(Collectors.toList());
List<``String``> list2 = stream.collect(Collectors.toCollection(ArrayList::new));
Set set1 = stream.collect(Collectors.toSet());
Stack stack1 = stream.collect(Collectors.toCollection(Stack::new));
// 3. String
String str = stream.collect(Collectors.joining()).toString();

一个 Stream 只可以使用一次,上面的代码为了简洁而重复使用了数次。


流的操作

接下来,当把一个数据结构包装成 Stream 后,就要开始对里面的元素进行各类操作了。常见的操作可以归类如下。

  • Intermediate:

map (mapToInt, flatMap 等)、 filter、 distinct、 sorted、 peek、 limit、 skip、 parallel、 sequential、 unordered

  • Terminal:

forEach、 forEachOrdered、 toArray、 reduce、 collect、 min、 max、 count、 anyMatch、 allMatch、 noneMatch、 findFirst、 findAny、 iterator

  • Short-circuiting:

anyMatch、 allMatch、 noneMatch、 findFirst、 findAny、 limit

这里不在举例每一个具体操作如何使用,详情可参考[1]

reduce(汇聚操作)

汇聚操作(也称为折叠)接受一个元素序列为输入,反复使用某个合并操作,把序列中的元素合并成一个汇总的结果。比如查找一个数字列表的总和或者最大值,或者把这些数字累积成一个List对象。Stream接口有一些通用的汇聚操作,比如reduce()和collect();也有一些特定用途的汇聚操作,比如sum(),max()和count()。注意:sum方法不是所有的Stream对象都有的,只有IntStream、LongStream和DoubleStream是实例才有。

java.util.stream.Collectors 类的主要作用就是辅助进行各类有用的 reduction 操作,例如转变输出为 Collection,把 Stream 元素进行归组。

下面会分两部分来介绍汇聚操作:

可变汇聚:把输入的元素们累积到一个可变的容器中,比如Collection或者StringBuilder;

其他汇聚:除去可变汇聚剩下的,一般都不是通过反复修改某个可变对象,而是通过把前一次的汇聚结果当成下一次的入参,反复如此。比如reduce,count,allMatch;

可变汇聚

可变汇聚对应的只有一个方法:collect,正如其名字显示的,它可以把Stream中的要有元素收集到一个结果容器中(比如Collection)。先看一下最通用的collect方法的定义(还有其他override方法):

1
<R> R collect(Supplier<R> supplier,BiConsumer<R, ? super T> accumulator,BiConsumer<R, R> combiner);

先来看看这三个参数的含义:Supplier supplier是一个工厂函数,用来生成一个新的容器;BiConsumer accumulator也是一个函数,用来把Stream中的元素添加到结果容器中;BiConsumer combiner还是一个函数,用来把中间状态的多个结果容器合并成为一个(并发的时候会用到)。看晕了?来段代码!

1
2
3
4
5
List<Integer> nums = Lists.newArrayList(1,1,null,2,3,4,null,5,6,7,8,9,10);
List<Integer> numsWithoutNull = nums.stream().filter(num -> num != null).
collect(() -> new ArrayList<Integer>(),
(list, item) -> list.add(item),
(list1, list2) -> list1.addAll(list2));

上面这段代码就是对一个元素是Integer类型的List,先过滤掉全部的null,然后把剩下的元素收集到一个新的List中。进一步看一下collect方法的三个参数,都是lambda形式的函数。

第一个函数生成一个新的ArrayList实例;

第二个函数接受两个参数,第一个是前面生成的ArrayList对象,二个是stream中包含的元素,函数体就是把stream中的元素加入ArrayList对象中。第二个函数被反复调用直到原stream的元素被消费完毕;

第三个函数也是接受两个参数,这两个都是ArrayList类型的,函数体就是把第二个ArrayList全部加入到第一个中;

但是上面的collect方法调用也有点太复杂了,没关系!我们来看一下collect方法另外一个override的版本,其依赖Collector

1
<R, A> R collect(Collector<? super T, A, R> collector);

这样清爽多了!Java8还给我们提供了Collector的工具类–Collectors,其中已经定义了一些静态工厂方法,比如:Collectors.toCollection()收集到Collection中, Collectors.toList()收集到List中和Collectors.toSet()收集到Set中。

这样的静态方法还有很多,这里就不一一介绍了,大家可以直接去看JavaDoc。下面看看使用Collectors对于代码的简化:

1
2
List<Integer> numsWithoutNull = nums.stream().filter(num -> num != null).
collect(Collectors.toList());

其它汇聚

– reduce方法:reduce方法非常的通用,后面介绍的count,sum等都可以使用其实现。reduce方法有三个override的方法,本文介绍两个最常用的。先来看reduce方法的第一种形式,其方法定义如下:

1
Optional<T> reduce(BinaryOperator<T> accumulator);

接受一个BinaryOperator类型的参数,在使用的时候我们可以用lambda表达式来。

1
2
List<Integer> ints = Lists.newArrayList(1,2,3,4,5,6,7,8,9,10);
System.out.println("ints sum is:" + ints.stream().reduce((sum, item) -> sum + item).get());

可以看到reduce方法接受一个函数,这个函数有两个参数,第一个参数是上次函数执行的返回值(也称为中间结果),第二个参数是stream中的元素,这个函数把这两个值相加,得到的和会被赋值给下次执行这个函数的第一个参数。要注意的是:第一次执行的时候第一个参数的值是Stream的第一个元素,第二个参数是Stream的第二个元素。这个方法返回值类型是Optional,这是Java8防止出现NPE的一种可行方法,后面的文章会详细介绍,这里就简单的认为是一个容器,其中可能会包含0个或者1个对象。

reduce方法还有一个很常用的变种:

1
T reduce(T identity, BinaryOperator<T> accumulator);

这个定义上上面已经介绍过的基本一致,不同的是:它允许用户提供一个循环计算的初始值,如果Stream为空,就直接返回该值。而且这个方法不会返回Optional,因为其不会出现null值。下面直接给出例子,就不再做说明了。

1
2
List<Integer> ints = Lists.newArrayList(1,2,3,4,5,6,7,8,9,10);
System.out.println("ints sum is:" + ints.stream().reduce(0, (sum, item) -> sum + item));

– count方法:获取Stream中元素的个数。比较简单,这里就直接给出例子,不做解释了。

– allMatch:是不是Stream中的所有元素都满足给定的匹配条件

– anyMatch:Stream中是否存在任何一个元素满足匹配条件

– findFirst: 返回Stream中的第一个元素,如果Stream为空,返回空Optional

– noneMatch:是不是Stream中的所有元素都不满足给定的匹配条件

– max和min:使用给定的比较器(Operator),返回Stream中的最大|最小值

高级用法如下案例:重点是看范型与函数式编程的结合

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
//实现双map转换
public static <K1,K2,T,U> Map<K1,Map<K2,U>> toNestMap(Collection<T> c,Function<T,K1> keyExe1,
Function<T,U> keyExe2,
Function<U,K2> keyExe3){
return c.stream().collect(Collectors.groupingBy(keyExe1::apply,
Collectors.collectingAndThen(Collectors.mapping(keyExe2::apply,Collectors.toList()),
list -> list.stream().collect(Collectors.toMap(keyExe3::apply,Function.identity(),(k1,k2) -> k1)))
));
}

//数据模型
public class ChannelMetadataObjectDTO {
private String metadataKey;
private ChannelMetadataDTO channelMetadataDTO;
}
public class ChannelMetadataDTO {
private String yqnCode;
private String yqnName;
private String yqnExtends;
private String targetCode;
private String targetName;
private String targetExtends;
}

//调用方式
List<ChannelMetadataObjectDTO> metadataMap
toNestMap(metadataMap,ChannelMetadataObjectDTO::getMetadataKey,ChannelMetadataObjectDTO::getChannelMetadataDTO,ChannelMetadataDTO::getYqnCode)

其它操作简要介绍如下

  • distinct: 对于Stream中包含的元素进行去重操作(去重逻辑依赖元素的equals方法),新生成的Stream中没有重复的元素;
  • filter: 对于Stream中包含的元素使用给定的过滤函数进行过滤操作,新生成的Stream只包含符合条件的元素;
  • map: 对于Stream中包含的元素使用给定的转换函数进行转换操作,新生成的Stream只包含转换生成的元素。这个方法有三个对于原始类型的变种方法,分别是:mapToInt,mapToLong和mapToDouble。这三个方法也比较好理解,比如mapToInt就是把原始Stream转换成一个新的Stream,这个新生成的Stream中的元素都是int类型。之所以会有这样三个变种方法,可以免除自动装箱/拆箱的额外消耗;
  • flatMap:和map类似,不同的是其每个元素转换得到的是Stream对象,会把子Stream中的元素压缩到父集合中;
  • peek: 生成一个包含原Stream的所有元素的新Stream,同时会提供一个消费函数(Consumer实例),新Stream每个元素被消费的时候都会执行给定的消费函数;
  • limit: 对一个Stream进行截断操作,获取其前N个元素,如果原Stream中包含的元素个数小于N,那就获取其所有的元素;
  • skip: 返回一个丢弃原Stream的前N个元素后剩下元素组成的新Stream,如果原Stream中包含的元素个数小于N,那么返回空Stream;

整体调用例子:

1
2
List<Integer> nums = Lists.newArrayList(1,1,null,2,3,4,null,5,6,7,8,9,10);
System.out.println(“sum is:”+nums.stream().filter(num -> num != null).distinct().mapToInt(num -> num * 2).peek(System.out::println).skip(2).limit(4).sum());

这段代码演示了上面介绍的所有转换方法(除了flatMap),简单解释一下这段代码的含义:给定一个Integer类型的List,获取其对应的Stream对象,然后进行过滤掉null,再去重,再每个元素乘以2,再每个元素被消费的时候打印自身,在跳过前两个元素,最后去前四个元素进行加和运算(解释一大堆,很像废话,因为基本看了方法名就知道要做什么了。这个就是声明式编程的一大好处!)。

大家可以参考上面对于每个方法的解释,看看最终的输出是什么。

2

4

6

8

10

12

sum is:36

可能会有这样的疑问:在对于一个Stream进行多次转换操作,每次都对Stream的每个元素进行转换,而且是执行多次,这样时间复杂度就是一个for循环里把所有操作都做掉的N(转换的次数)倍啊。其实不是这样的,转换操作都是lazy的,多个转换操作只会在汇聚操作(见下节)的时候融合起来,一次循环完成。我们可以这样简单的理解,Stream里有个操作函数的集合,每次转换操作就是把转换函数放入这个集合中,在汇聚操作的时候循环Stream对应的集合,然后对每个元素执行所有的函数。


进阶:自己生成流

Stream.generate

通过实现 Supplier 接口,你可以自己来控制流的生成。把 Supplier 实例传递给 Stream.generate() 生成的 Stream,默认是串行(相对 parallel 而言)但无序的(相对 ordered 而言)。由于它是无限的,在管道中,必须利用 limit 之类的操作限制 Stream 大小。

生成 10 个随机整数
1
2
3
4
5
6
Random seed = new Random();
Supplier<``Integer``> random = seed::nextInt;
Stream.generate(random).limit(10).forEach(System.out::println);
//Another way
IntStream.generate(() -> (int) (System.nanoTime() % 100)).
limit(10).forEach(System.out::println);

Stream.generate() 还接受自己实现的 Supplier。例如在构造海量测试数据的时候,用某种自动的规则给每一个变量赋值;或者依据公式计算 Stream 的每个元素值。这些都是维持状态信息的情形。

自实现 Supplier
1
2
3
4
5
6
7
8
9
10
11
12
Stream.generate(new PersonSupplier()).
limit(10).
forEach(p -> System.out.println(p.getName() + ", " + p.getAge()));

private class PersonSupplier implements Supplier<Person> {
private int index = 0;
private Random random = new Random();
@Override
public Person get() {
return new Person(index++, "StormTestUser" + index, random.nextInt(100));
}
}

输出结果:

1
2
3
4
5
6
7
8
9
10
StormTestUser1, 9
StormTestUser2, 12
StormTestUser3, 88
StormTestUser4, 51
StormTestUser5, 22
StormTestUser6, 28
StormTestUser7, 81
StormTestUser8, 51
StormTestUser9, 4
StormTestUser10, 76

Stream.iterate

iterate 跟 reduce 操作很像,接受一个种子值,和一个 UnaryOperator(例如 f)。然后种子值成为 Stream 的第一个元素,f(seed) 为第二个,f(f(seed)) 第三个,以此类推。

生成一个等差数列

1
Stream.iterate(0, n -> n + 3).limit(10). forEach(x -> System.out.print(x + " "));.

输出结果:

1
0 3 6 9 12 15 18 21 24 27

与 Stream.generate 相仿,在 iterate 时候管道必须有 limit 这样的操作来限制 Stream 大小。


结束语

  • 不是数据结构

  • 它没有内部存储,它只是用操作管道从 source(数据结构、数组、generator function、IO channel)抓取数据。

  • 它也绝不修改自己所封装的底层数据结构的数据。例如 Stream 的 filter 操作会产生一个不包含被过滤元素的新 Stream,而不是从 source 删除那些元素。

  • 所有 Stream 的操作必须以 lambda 表达式为参数

  • 不支持索引访问

  • 惰性化(即intermediate操作)

  • 很多 Stream 操作是向后延迟的,一直到它弄清楚了最后需要多少数据才会开始。

  • 并行能力

  • 当一个 Stream 是并行化的,就不需要再写多线程代码,所有对它的操作会自动并行进行的。

  • 可以是无限的

    • 集合有固定大小,Stream 则不必。limit(n) 和 findFirst() 这类的 short-circuiting 操作可以对无限的 Stream 进行运算并很快完成。


参考内容:

[1]https://www.ibm.com/developerworks/cn/java/j-lo-java8streamapi/


版权声明:本文为博主原创文章,欢迎转载,转载请注明作者、原文超链接,感谢各位看官!!!

本文出自:monkeyGeek

座右铭:生于忧患,死于安乐

欢迎志同道合的朋友一起交流、探讨!

monkeyGeek
#

评论

Your browser is out-of-date!

Update your browser to view this website correctly. Update my browser now

×