第四章：函数式数据处理(一)-----流（Stream）-白红宇

本文是学习Java8,参考JAVA8 IN ACTION这本书，学习整理以及自己的总结,推荐这本书；

1:引入流

流(Stream)是javaAPI的新成员,它允许你以声明性方式处理数据集(通过查询语句来表达而不是临时编写一个实现). 此外,流还可以并行的进行处理,你无须写任何多线程代码了.

首先,我们以一个例子看下流的使用:下面两段代码都是用来返回低热量的菜肴名称的，并按照卡路里排序，一个是用Java 7写的，另一个是用Java 8的流写的。比较一下。不用太担心 Java 8代码怎么写，我们在接下来的几节里会详细解释。首先新建一个实体类Dish.java

@Data@AllArgsConstructor@ToStringpublic class Dish {    private final String name;    private final boolean vegetarian;    private final int calories;    private final Type type;        public enum Type { MEAT, FISH, OTHER }        public static final List
    
      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));}复制代码

Java8之前的操作：

/**     * java7     */    @Test    public  void testJava7(){        ArrayList
    
      lowcaloriesDishs=new ArrayList<>();        //筛选出低卡路里的菜肴        for (Dish dish:list){            if (dish.getCalories()<400) {                lowcaloriesDishs.add(dish);            }        }        //按照卡路里进行排序        Collections.sort(lowcaloriesDishs, new Comparator
     
      () {            @Override            public int compare(Dish o1, Dish o2) {                //升序              //  return Integer.compare(o1.getCalories(),o2.getCalories());                //降序                return Integer.compare(o2.getCalories(),o1.getCalories());            }        });        //输出低卡路里的菜品        for (Dish dish:lowcaloriesDishs){            System.out.println(dish.getName()+":"+dish.getCalories());        }    }复制代码

得到结果:season fruit:120rice:350复制代码

在这段代码中，你用了一个“垃圾变量” lowcaloriesDishs 。它唯一的作用就是作为一次性的中间容器。在Java 8中，实现的细节被放在它本该归属的库里了。

使用Java8的操作：

/**     * java8     */    @Test    public void test2(){        List
    
      lowDishs=list.stream()                //筛选出低于400的食物                .filter(a->a.getCalories()<400)                // .sorted((a,b)->b.getCalories()-a.getCalories())                //排序                 .sorted(Comparator.comparing(Dish::getCalories))                //输出菜肴名称                 .map(Dish::getName)                 .collect(toList());        System.out.println(lowDishs);    }复制代码

输出的结果是:[season fruit, rice]复制代码

为了利用多核架构并行执行这段代码，你只需要把 stream() 换成 parallelStream() ：

/**     * java8 parallelStream     * java8多核架构并行执行这段代码     */    @Test    public void test3(){        List
    
      lowDishs=list.parallelStream()                .filter(a->a.getCalories()<400)                .sorted(Comparator.comparing(Dish::getCalories))                .map(Dish::getName)                .collect(toList());        System.out.println(lowDishs);    }复制代码

同样得到结果：[season fruit, rice]复制代码

你可能会想，在调用 parallelStream 方法的时候到底发生了什么。用了多少个线程？对性能有多大提升？后面会详细讨论这些问题

现在，你可以看出，从软件工程师的角度来看，新的方法有几个显而易见的好处:

代码是以声明性方式写的:说明想要完成什么,而不是说明如何实现一个操作(利用for if 等控制语句).这种方法加上行为参数化,可以让你很轻松的应对变化的需求,你很容易再创建一个代码版本，利用 Lambda表达式来筛选高卡路里的菜肴，而用不着去复制粘贴代码

你可以把几个基础操作连接起来:来表达复杂的数据流水线工作,同时保证代码清晰可读.filter 的结果被传给了 sorted 方法，再传给 map 方法，最后传给 collect 方法。

需要注意的是: filter(),sorted(),map(), 返回的都是流(Stream),都是Stream的方法,collect()方法除外.

2：流的简介

java8中的集合支持一个新的stream()方法,它会返回一个流,接口定义在 java.util.stream.Stream中.

那么，流到底是什么呢？简短的定义就是“从支持数据处理操作的源生成的元素序列”。让我们一步步剖析这个定义:

元素序列: 就像集合一样,流提供了一个接口,可以访问特定元素类型的一组有序值.因为集合是数据结构,所以它的主要目的是以特定的时间/空间复杂度来存储访问元素.但流的目的在于表达计算.

源: 流会使用一个提供数据的源,这些源可以是数组,集合,或输入输出资源.注意:从有序结合生成的流会保留原有的顺序,由列表生成的流,其元素顺序也与列表一致.

数据处理操作: 流的数据处理功能类似于数据库的操作.以及函数式编程语言的常用操作.如 filter 、 map 、 reduce 、 find 、 match 、 sort 等。流操作可以顺序执行，也可并行执行。此外，流操作有两个重要的特点。

流水线: 很多流操作本身会返回一个流.这样多个操作就可以连接起来形成一个更大的流水线.流水线操作可以看成对数据源进行数据库式查询.

内部迭代: 与使用迭代器对集合进行显示迭代不同,流的迭代都是在背后进行的.

让我们来看一段能够体现所有这些概念的代码：

@Test    public void test4() {        List
    
      lowCaloricDishesName =                //1.从 menu 获得流（菜肴列表）,建立操作流水线                menu.parallelStream()                        //2.选出高热量菜肴                        .filter(d -> d.getCalories() > 300)                        //3.输出菜肴名称                        .map(Dish::getName)                        //4.只选择前三个                        .limit(3)                        //5.将结果保存在另一个List中                        .collect(toList());        System.out.println(lowCaloricDishesName);    }复制代码

运行得到结果：[rice, chicken, prawns]复制代码

在本例中,我们显示对menu进行stream操作,得到一个流,数据源是菜肴列表menu,接下来对流进行一系列数据处理操作:filter 、 map 、 limit 和 collect 。除了 collect 之外，所有这些操作都会返回另一个流，这样它们就可以接成一条流水线,于是就可以看作对源的一个查询. 最后collect开始处理流水线,并返回一个结果(collect和别的操作不一样,它返回的不是一个流而是一个list). 在调用collect之前,没有任何结果产生,事实上,根本就没有从menu里选择元素.你可以这么理解：链中的方法调用都在排队等待，直到调用 collect

图4-2显示了流操作的顺序： filter 、 map 、 limit 、 collect ，每个操作简介如下。

在进一步介绍能对流做什么操作之前，先让我们回过头来看看Collection API和新的Stream API的思想有何不同

3：集合与流

粗略的讲,流与集合的差异就在于什么时候进行计算,集合是内存中的数据结构,它包含数据结构中目前所有的值(结合中每个元素必须先计算出来才能添加到集合中.) （你可以往集合里加东西或者删东西，但是不管什么时候，集合中的每个元素都是放在内存里的，元素都得先算出来才能成为集合的一部分。）

相比之下,流是再概念上固定的数据结构.这个思想就是用户仅仅从流中提取需要的值,而这些值,在用户看不见的地方,只会按需生成. 这是一种生产者--消费者的关系,从另一个角度来说,流就想一个延迟创建的集合:只有在消费者要求的时候才会计算值。与此相反，集合则是急切创建的。

4：流-只能遍历一次

请注意,和迭代器一样,流只能遍历一次,遍历完之后,我们就说这个流已经被消费掉了, 你可以从原始数据源那里再获得一个新的流来重新遍历一遍，就像迭代器一样（这里假设它是集合之类的可重复的源，如果是I/O通道就没戏了）。

例如，以下代码会抛出一个异常，说流已被消费掉了：java.lang.IllegalStateException: stream has already been operated upon or closed

/**     * java8多次使用流异常     */    @Test    public void test5(){        List
    
      title = Arrays.asList("Java8", "In", "Action");        Stream
     
       stream=title.stream();        stream.forEach(a-> System.out.println(a));        //这种情况下，会报错：java.lang.IllegalStateException: stream has already        // been operated upon or closed        //因为流已经关闭了，只能使用一次，要想使用，重新获取流        stream.forEach(b-> System.out.println(b));    }复制代码

所以要记得，流只能消费一次！

5:内部迭代与外部迭代

使用Collection接口需要用户做迭代（eg:for-each）,这称为外部迭代；相反，Streams库使用内部迭代---它不仅把迭代做了，还把得到的流的值存在某个地方，你只要给他一个函数说要干什么就可以了；相关区别请看一下代码：

内部迭代时，项目可以透明的并行处理，或者用更优化的顺序进行处理，，这差不多就是Java8引入流的理由了---Streams库的内部迭代可以自动选择一种合适你硬件的数据表示和并行实现

6：流操作

流操作可以分为中间操作和终端操作； java.util.stream.Stream 中的 Stream 接口定义了许多操作。它们可以分为两大类. 可以连接起来的流操作称为中间操作，关闭流的操作称为终端操作下图展示了这两类操作：

6.1：中间操作

像filter,map,sort等中间操作会返回一个流；我们把代码改一下，让每一步操作都返回一个当前处理的值；然后对比一下结果

/**     * 对比每一步的输出     */    @Test    public  void  test6(){        List
    
      names=list.stream()                .filter(a->{                    System.out.println("filter:"+a.getName());                    return a.getCalories()>300;                })                .map(d->{                    System.out.println("map:"+d.getName());                     return d.getName();                })                .limit(3)                .collect(toList());        System.out.println("names:"+names);    }复制代码

输出的结果：

filter:porkmap:porkfilter:beefmap:beeffilter:chickenmap:chickennames:[pork, beef, chicken]复制代码

是不是和我们预想的不同，这种操作充分利用了流的延迟特性，尽管filter和map是两个独立的操作，但是他们合并到同一次遍历中了，称为循环合并；

如果我们将上面的代码中的stream换成并行流 parallelStream，看一下运行结果有什么不同

filter:pizzafilter:porkmap:porkfilter:prawnsfilter:salmonfilter:ricemap:ricefilter:season fruitfilter:beefmap:beeffilter:chickenmap:chickenfilter:french friesmap:french friesmap:salmonmap:prawnsmap:pizzanames:[pork, beef, chicken]复制代码

每次运行结果每一步骤的结果也是不同的；

6.2：终端操作

终端操作会从流中生成结果，其结果不是任何流中的值，比如：List,Integer,甚至是void; 例如在下面的流水线中,foerach是一个返回void的终端操纵,它会对源中的每道菜应用一个Lambda,把 System.out.println 传递给 forEach ，并要求它打印出由 menu 生成的流中的每一个 Dish ：

menu.stream().forEach(System.out::println);复制代码

7:小结

流是“从支持数据处理操作的源生成的一系列元素”

流利用内部迭代：迭代通过 filter 、 map 、 sorted 等操作被抽象掉了。

流操作有两类：中间操作和终端操作。

filter 和 map 等中间操作会返回一个流，并可以链接在一起。可以用它们来设置一条流水线，但并不会生成任何结果。

forEach 和 count 等终端操作会返回一个非流的值，并处理流水线以返回结果流中的元素是按需计算的