Java 8 stream学习

简介

    Java 8里引入的另外一个重要特性就是stream api。笼统的来说，它这种特性的引入可以方便我们以一种更加声明式的方式来写代码，更加便利了一些函数式编程方法的使用。同时，它也使得我们可以充分利用系统的并行能力而不用自己手工的去做很多底层的工作。当然，里面最让人印象深刻的也许是一种类似于流式编程的概念。 

 

流水线(pipeline)

    在以前一些linux脚本命令中经常会接触到的一个概念就是pipeline，它其实体现出来了一个很好的程序设计哲学，就是我们应该设计很多小而且职责单一的模块。每个模块只专注于做一件事情，然后它们之间通过一种流水线的方式将它们连接起来。我们看一个典型的命令： 

cat file1 file2 | tr "[A-Z]" "[a-z]" | sort | tail -3

    上面这部分的命令表示从file1, file2两个文件里读内容，然后将里面的大写字母转换成小写字母，然后再排序。最后取排序后的最后3个字符。

    我们这里其实不是关心这个命令做了什么，而是这些命令它们执行的方式。实际上，在linux里面，上述的几个命令它们完全是并发执行的，前面的cat命令可能是读取了一部分文件的内容经由tr命令替换字符后，再由sort命令排序。它们的执行过程如下图所示：

     上述的执行过程类似于一个工厂里的生产流水线，在每个生产的步骤里，它不是等前面一个步骤要生产的所有东西都完成才做下一步，而是前面做完一部分就马上传递给后面一个部分。这样才能实现所有步骤的并发工作。如果熟悉python的同学，也许会联想到里面的generator的功能，它的功能也是类似的。

    那么，上述的这种流水线式的编程方式有什么好处呢？除了前面提到的它可以使得我们充分利用计算机的并发能力，还能够处理一些数据量很大的场景。因为它不是所有的数据都要一次性的放到内存里来处理。另外，它的每个步骤如果定义好之后，确实可以结合前面函数式编程的讨论得到一个很好的应用。

    现在，java 8里面引入的stream特性，就是给我们带来了上述的好处。我们来详细分析一下。

 

示例对比

    假设我们有一个如下类：

import java.util.*;

public class Dish {

    private final String name;
    private final boolean vegetarian;
    private final int calories;
    private final Type type;

    public Dish(String name, boolean vegetarian, int calories, Type type) {
        this.name = name;
        this.vegetarian = vegetarian;
        this.calories = calories;
        this.type = type;
    }

    public String getName() {
        return name;
    }

    public boolean isVegetarian() {
        return vegetarian;
    }

    public int getCalories() {
        return calories;
    }

    public Type getType() {
        return type;
    }

    public enum Type { MEAT, FISH, OTHER }

    @Override
    public String toString() {
        return name;
    }

    public static final List<Dish> menu =
            Arrays.asList( new Dish("pork", false, 800, Dish.Type.MEAT),
                           new Dish("beef", false, 700, Dish.Type.MEAT),
                           new Dish("chicken", false, 400, Dish.Type.MEAT),
                           new Dish("french fries", true, 530, Dish.Type.OTHER),
                           new Dish("rice", true, 350, Dish.Type.OTHER),
                           new Dish("season fruit", true, 120, Dish.Type.OTHER),
                           new Dish("pizza", true, 550, Dish.Type.OTHER),
                           new Dish("prawns", false, 400, Dish.Type.FISH),
                           new Dish("salmon", false, 450, Dish.Type.FISH));
}

    这个示例稍微有点长，主要是定义了一个Dish对象，然后初始化了一个Dish的list。 

    现在假设我们需要做一些如下的操作，首先获取列表里卡路里小于400的元素，然后再根据卡路里的数值进行排序，最后我们再返回这些排序后的元素的名字。如果按照我们通常的理解，会做一个如下的实现：

List<Dish> lowCaloricDishes = new ArrayList<>();
for(Dish d: menu) {
    if(d.getCalories() < 400) {
        lowCaloricDishes.add(d);
    }
}
Collections.sort(lowCaloricDishes, new Comparator<Dish>() {
    public int compare(Dish d1, Dish d2) {
        return Integer.compare(d1.getCalories(), d2.getCalories());  
    }
});
List<String> lowCaloricDishesName = new ArrayList<>();
for(Dish d: lowCaloricDishes) {
    lowCaloricDishesName.add(d.getName());
}

    上面这部分的代码看起来很中规中矩，当然，也显得有点啰嗦。具体它的特点以及与后面的代码对比会在后面详细说。如果我们用stream api来实现上述的逻辑该怎么做呢？

 

import static java.util.Comparator.comparing;
import static java.til.stream.Collectors.toList;

List<String> lowCaloricDishesName = menu.stream()
            .filter(d -> d.getCalories < 400)
            .sorted(comparing(Dish::getCalories))
            .map(Dish::getName)
            .collect(toList());

     现在我们来详细比较一下两种写法上的差别。在第一种写法上，我们需要过滤数据元素的时候需要使用一个临时的list来保存过滤后的结果，然后再将过滤后的元素排序。因为我们最后需要的是一个排序后元素的名字列表，于是没办法，又要创建一个list，将里面的元素一个个的获取出来再填充到这个list里。所以综合来说，这种方法需要创建大量临时的列表。这样不但使得程序变得冗长难懂，而且创建的这些临时的列表也增加了程序垃圾回收的压力。

    我们再看stream api的实现方式。上述代码的实现更加是声明式的，它的处理流程更加像一个流水线的方式。我们首先利用filter方法来过滤元素，然后调用sorted方法来排序，最后用map方法来转换提取的元素。这种写法不仅更加简洁而且更加高效。关于这些具体方法的意思我们在后续部分详细讨论。

 

stream定义

    从前面使用手法上来看，stream的使用像是一个流水线。在这个流水线里，它好像有一个自动推进的流程，我们使用者只需要指定对它的各种转换操作就可以了。从更严格的意义来说，stream是一组定义的计算序列，这种结构将一系列的操作给串联起来。所以如果熟悉设计模式的人会觉得这就像是一个chain of responsibility模式。当然，从函数式编程的理论角度来说，它表示的是一个叫monad的结构。

    因此，从定义的角度来说，stream定义的并不是一个普通意义上的数据流，它实际上是一个计算流，表示一组计算的顺序。它有一些典型的特性，比如内循环(internal iteration), 惰性计算(laziness)等。我们结合它们和集合类的比较来一起讨论。

 

内迭代和外迭代(internal iteration vs external iteration)

    在前面示例代码里，我们已经比对过两种实现方法，对于第一种方法来说，它需要显式的定义一个循环迭代的过程。比如：

for(Dish d: menu) {
    if(d.getCalories() < 400) {
        lowCaloricDishes.add(d);
    }
}

     这部分代码的本质是集合实现了一个iterable的接口，然后在这个循环里调用iterator()方法这样依次的遍历集合里的元素。这种方式实现的代码有如下几个问题：

1. for循环本身就是串行的过程，所有集合里元素处理的顺序必须按照定义好的顺序来处理。

2. 因为这种循环是由开发人员来写的，而不是本身库内部定义的，这样系统比较难做一些内在的优化，比如数据的重排序，潜在并行性的利用等。

    尤其是牵涉到大量数据和性能的时候，如果有更加好的方式来优雅的处理程序逻辑将更加受到欢迎。

    与前面对应的是另外一种遍历方式，称为内部迭代。和上述代码对应的一种实现如下：

menu.stream()
            .filter(d -> d.getCalories < 400)

     从语法上看起来它只是一个很小的变化，但是它的实际实现却是差别很大的。因为这里的代码并没有显式的定义循环处理的过程，真正迭代处理的过程相当于交给类库来处理了。类库的实现可以潜在的利用一些优化的手段来使得程序的执行性能更加高效。所以一旦看到stream的时候，对它执行运算时就好像已经在一个生产线的传送带上了。所有需要做的事情就是将一些具体的操作传递给这个流水线。

    前面这种方式的实现实际上将要做什么和怎么做是混在一起的。比如说我需要过滤出来所有卡路里小于400的菜，这里就需要循环遍历所有的菜列表。而后面的这种方式更像是一个声明，只是说我需要过滤某个东西。而这个东西的条件就是一个lambda表达式，至于它的过滤是怎么实现的我们可以不用去关心了。 这样整个业务逻辑的代码实现也更加清晰简练。

 

stream工作方式

不可变性

    基于前面的示例，我们可能有若干个疑问，因为前面按照传统的方式来实现的功能需要用到临时的列表，必然要修改一些元素的属性。那么在stream里面，我们调用的那些处理方法它会不会修改原有stream数据源的值呢？我们看如下的代码：

 

List<String> myList = new ArrayList<>();
myList.add("a1");
myList.add("a2");
myList.add("b1");
myList.add("c2");
myList.add("c1");
myList.stream()
    .filter(s -> s.startsWith("a"))
    .map(String::toUpperCase)
    .sorted()
    .forEach(System.out::println);
System.out.println(myList);

   它的输出如下：

A1
A2
[a1, a2, b1, c2, c1]

    上述代码里的filter, map等方法并没有修改stream里源的内容。它仅仅是根据当前的转换操作新建一个元素。这种思路恰恰也是和copy on write的数据结构暗合的。而且它对于以后的并发处理也是有巨大的好处。

 

不可重复使用

    stream还有一个典型的特征就是它不能被重复使用，比如说我们尝试如下的代码：

 Stream<String> stream = myList.stream();

stream.anyMatch(s -> true);
stream.anyMatch(s -> true);

   在编译的时候没有问题，而运行的时候将出现如下的错误：

 

Exception in thread "main" java.lang.IllegalStateException: stream has already been operated upon or closed
	at java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:229)
	at java.util.stream.ReferencePipeline.anyMatch(ReferencePipeline.java:449)
	at Sample.main(Sample.java:20)

    因此，凡是我们使用的stream它就相当于一次性的用品，用完之后就会被close了。如果我们需要再利用stream进一步的操作需要重新声明一个新的stream。

 

两种运算

    在前面的代码里还要一个需要我们深入了解的地方就是，我们能够对一个stream做哪些操作呢？像前面的filter, map, forEach, collect等。它们有什么作用呢？

    在stream里，主要有两种运算，一种叫中间运算(intermediate)，还要一种是终止运算(terminal)。比如前面的filter, map运算。filter运算仅仅过滤stream里的元素，但是返回的依然是一个Stream<String>类型。同样，map操作也仅仅实现一个元素的转换。如果我们有一些类型转换的话，实际上也只是将一种类型参数的Stream转换成另外一种Stream。而终止运算比如前面的collect，它将一个Stream又转换成了一个List，类似的它还要toSet等方法。这些方法使得stream的处理终止。所以我们称之为终止运算方法。关于intermediate和terminal方法的详细介绍可以参考Stream的官方文档，如下链接。

  

惰性计算(laziness)

    stream里还要一个比较典型的特性就是惰性计算。像前面stream里的一些典型运算filter, mapping。它们可以通过急性求值的方式来实现。以filter方法为例，也就是说在方法返回前，急性求值就需要完成对所有数据元素的过滤。而惰性计算则是当需要的时候才进行过滤运算。在实际应用中，惰性计算的方式更加有优势一些。因为我们将它和流水线后续的一些操作结合在一起来运算，而不用多次遍历数据。在某些场景里，比如说我们需要遍历一个非常大的集合去寻找第一个匹配某种条件的数据，我们完全可以在找到第一个匹配的元素时就返回，而不用真正去完整遍历整个集合。这种特性尤其在数据集合有无限个长度的情况下用处比较明显。

    我们来看一个如下的示例：

Stream.of("d2", "a2", "b1", "b3", "c")
    .filter(s -> {
        System.out.println("filter: " + s);
        return true;
    });

    上述的stream操作里只有一个filter操作，相当于只是做了一个stream转换成另外一个stream的操作，并没有一个terminal的操作。如果运行上面的代码的话，则不会有任何输出。

    总的来说，对于一个stream的操作它会尽量采用惰性计算的方式以实现满足目标结果。

 

stream执行顺序

    还有一个比较值得让人关心的就是stream处理元素的执行顺序。它是按照前面示例里某个运算一次将所有的数据处理完之后再传递给下一个呢还是一次处理一个传递下去呢？我们再来看如下的代码：

Stream.of("d2", "a2", "b1", "b3", "c")
    .filter(s -> {
        System.out.println("filter: " + s);
        return true;
    })
    .forEach(s -> System.out.println("forEach: " + s));

     运行上面这部分程序的输出如下：

filter: d2
forEach: d2
filter: a2
forEach: a2
filter: b1
forEach: b1
filter: b3
forEach: b3
filter: c
forEach: c

    可见，在stream里对元素的处理是按照流水线的方式来进行的。因此它不需要额外的利用集合的数据结构来保存中间结果。这种方式在处理海量数据的时候带来非常遍历的特性。

 

Optional类型

    Stream api带来的另外一个影响就是引入了optional类型的数据。关于optional类型数据的详细讨论会在后面的文章里描述。这里只是一个简单的叙述。我们来看如下的示例：

Optional<Shape> firstBlue = shapes.stream()
        .filter(s -> s.getColor() == BLUE)
        .findFirst();

    在shapes的stream里通过filter方法来过滤一个符合color == BLUE的元素。实际上返回的结果可能存在有这样的元素，也可能不存在这样的元素。于是针对这种可能存在也可能不存在的类型元素，这里引入了Optional类型数据来描述它。通过引入Optional类型可以减少和规避很多容易出现nullpointerexception的情况。也算是对程序的一种改进。

 

stream的潜在并行性

    前面提到过，在stream api里引入了一种使得运用并行开发更加简便的方式，这就是 parallel stream。在目前多核体系结构比较普遍的情况下，大多数计算机都有多个核，如果只是使用以前的编程方式的话并不能充分发挥机器的性能。于是需要一种更好的方式来使用多核和多线程。在以往的java 多线程开发里，使用好多线程是一个很困难的任务。于是为了简化对一些多线程情况下的使用，这里就引入了parallel stram。

    需要注意的是，前面用的stream是对数据进行串行处理的，而这里使用并行处理的时候，它的使用方式则稍微有点差别。我们先来看一部分如下的代码：

Arrays.asList("a1", "a2", "b1", "c2", "c1")
            .parallelStream()
            .filter(s -> {
                System.out.format("filter: %s [%s]\n",
                    s, Thread.currentThread().getName());
                return true;
            })
            .map(s -> {
                System.out.format("map: %s [%s]\n",
                    s, Thread.currentThread().getName());
                return s.toUpperCase();
            })
            .forEach(s -> System.out.format("forEach: %s [%s]\n",
                s, Thread.currentThread().getName()));

    这部分代码看起来比较复杂，实际上和前面代码的唯一差别就是stream()方法编程了parallelStream()。在每个处理步骤里都加入了打印的消息以方便我们跟踪程序执行的过程。如果我们运行上述的代码，会发现如下的输出：

 

filter: b1 [main]
map: b1 [main]
filter: c2 [ForkJoinPool.commonPool-worker-4]
filter: c1 [ForkJoinPool.commonPool-worker-3]
map: c1 [ForkJoinPool.commonPool-worker-3]
forEach: C1 [ForkJoinPool.commonPool-worker-3]
filter: a2 [ForkJoinPool.commonPool-worker-1]
map: a2 [ForkJoinPool.commonPool-worker-1]
forEach: A2 [ForkJoinPool.commonPool-worker-1]
filter: a1 [ForkJoinPool.commonPool-worker-2]
map: c2 [ForkJoinPool.commonPool-worker-4]
forEach: C2 [ForkJoinPool.commonPool-worker-4]
forEach: B1 [main]
map: a1 [ForkJoinPool.commonPool-worker-2]
forEach: A1 [ForkJoinPool.commonPool-worker-2]

    实际上，如果我们多次运行程序的话会发现每次的输出还有点不一样。当然，从输出里我们还可以看到一个东西，就是输出的线程名是属于一个ForkJoinPool里的线程。也就是说它实际上运用了线程池。这里运用到的线程池就是java 7里引入的forkjoin pool。关于forkjoin pool的讨论可以参考我前面一篇相关的分析文章。

 

总结

    Java 8 引入的stream api可以说是给前面函数式编程应用到的lambda表达式提供了一个极好的应用场景。它本质上是一个惰性计算流，它不像我们传统使用的数据结构，需要事先分配内存空间，而是一种按需计算的模式。所以它更像是一个流水线式的计算模型。同时，它在默认的情况下是串行执行的，所以它的执行顺序不一样，但是可以利用很少的内存空间。另外，在stream api里也有很简单支持并行计算的parallemstream，它本质上是运用了java的Forkjoin thread pool来实现并行的。这种方式大大简化了我们并发编程的难度。

 

参考材料

http://cr.openjdk.java.net/~briangoetz/lambda/lambda-libraries-final.html

http://winterbe.com/posts/2014/07/31/java8-stream-tutorial-examples/

http://www.oracle.com/technetwork/articles/java/ma14-java-se-8-streams-2177646.html

http://www.oracle.com/technetwork/articles/java/architect-streams-pt2-2227132.html

http://www.amazon.com/Java-Action-Lambdas-functional-style-programming/dp/1617291994/ref=sr_1_1?s=books&ie=UTF8&qid=1447684950&sr=1-1&keywords=java+8+in+action