2020-08-09-Java之Stream流的使用总结

Java—Stream

什么是Stream？

Java8 中，Collection 新增了两个流方法，分别是 Stream() 和 parallelStream()

Java8 中添加了一个新的接口类 Stream，相当于高级版的 Iterator，它可以通过 Lambda 表达式对集合进行大批量数据操作，或 者各种非常便利、高效的聚合数据操作。

为什么要使用 Stream？

在 Java8 之前，我们通常是通过 for 循环或者 Iterator 迭代来重新排序合并数据，又或者通过重新定义 Collections.sorts 的 Comparator 方法来实现，这两种方式对于大数据量系统来说，效率并不是很理想。Stream 的聚合操作与数据库 SQL 的聚合操作 sorted、filter、map 等类似。我们在应用层就可以高效地实现类似数据库 SQL 的 聚合操作了，而在数据操作方面，Stream 不仅可以通过串行的方式实现数据操作，还可以通过并行的方式处理大批量数据，提高数据 的处理效率。

Stream 使用入门

@Test
  public void test1(){
      List<String> names = Arrays.asList("张三","李四","王五","赵柳","张五六七","王少","赵四","张仁","李星");
      //需求：找出 姓张中名字最长的
      int maxLengthStartWithZ = names.parallelStream()
              .filter(name -> name.startsWith("张"))
              .mapToInt(String::length)
              .max()
              .getAsInt();
      System.out.println(names.get(maxLengthStartWithZ));
  }

Stream 操作分类

官方将 Stream 中的操作分为两大类：终结操作（Terminal operations）和中间操作（Intermediate operations）。

中间操作会返回一个新的流，一个流可以后面跟随零个或多个中间操作。其目的主要是打开流，做出某种程度的数据映射/过滤，

然后会返回一个新的流，交给下一个操作使用。这类操作都是惰性化的（lazy），就是说，仅仅调用到这类方法，并没有真正开始流的

遍历。而是在终结操作开始的时候才真正开始执行。

中间操作又可以分为无状态（Stateless）与有状态（Stateful）操作:

​ 无状态是指元素的处理不受之前元素的影响；

​ 有状态是指该操作只有拿到所有元素之后才能继续下去。

终结操作是指返回最终的结果。一个流只能有一个终结操作，当这个操作执行后，这个流就被使用“光”了，无法再被操作。所以

这必定这个流的最后一个操作。终结操作的执行才会真正开始流的遍历，并且会生成一个结果。

终结操作又可以分为短路（Short-circuiting）与非短路（Unshort-circuiting）操作，

​ 短路是指遇到某些符合条件的元素就可以得到最终结果，

​ 非短路是指必须处理完所有元素才能得到最终结果。操作分类详情如下图所示：

个别方法的测试

public class StuWithStream {
    public static void main(String[] args) {
        List<Student> studentList =Datainit();
        groupBy(studentList);
       // filter(studentList);
//        total(studentList);
//        MaxAndMin(studentList);

    }
    public static List<Student> Datainit(){
        List<Student> students = Arrays.asList(
                new Student("小明", 168, "男"),
                new Student("大明", 182, "男"),
                new Student("小白", 174, "男"),
                new Student("小黑", 186, "男"),
                new Student("小红", 156, "女"),
                new Student("小黄", 158, "女"),
                new Student("小青", 165, "女"),
                new Student("小紫", 172, "女"));
        return students;
    }
    //Stream实现分组
    public static  void groupBy(List<Student> studentsList){
        Map<String, List<Student>> groupBy = studentsList
                .stream()
                .collect(Collectors.groupingBy(Student::getSex));
        System.out.println("分组后："+groupBy);
    }

    //Stream实现过滤
    public static  void filter(List<Student> studentsList){
        List<Student> filter = studentsList
                .stream()
                .filter(student->student.getHeight()>180)
                .collect(Collectors.toList());
        System.out.println("过滤后："+filter);
    }

    //Stream实现求和
    public static  void total(List<Student> studentsList){
        int totalHeight = studentsList
                .stream()
                .mapToInt(Student::getHeight)
                .sum();
        System.out.println(totalHeight);
    }
    //Stream找最大和最小
    public static  void MaxAndMin(List<Student> studentsList){
        int maxHeight = studentsList
                .stream()
                .mapToInt(Student::getHeight)
                .max()
                .getAsInt();
        System.out.println("max:"+maxHeight);
        int minHeight = studentsList
                .stream()
                .mapToInt(Student::getHeight)
                .min()
                .getAsInt();
        System.out.println("min:"+minHeight);
    }



}

因为 Stream 操作类型非常多，总结一下常用的

• map():将流中的元素进行再次加工形成一个新流，流中的每一个元素映射为另外的元素。
• filter(): 返回结果生成新的流中只包含满足筛选条件的数据
• limit()：返回指定数量的元素的流。返回的是 Stream 里前面的 n 个元素。
• skip()：和 limit()相反，将前几个元素跳过（取出）再返回一个流，如果流中的元素小于或者等于 n，就会返回一个空的流。
• sorted()：将流中的元素按照自然排序方式进行排序。
• distinct()：将流中的元素去重之后输出。
• peek()：对流中每个元素执行操作，并返回一个新的流，返回的流还是包含原来流中的元素。

Stream 并发 Stream 源码实现

并发的处理函数对比 parallelStream()方法这里的并行处理指的是，Stream 结合了 ForkJoin 框架，对 Stream 处理进行了分片，Splititerator 中estimateSize 方法会估算出分片的数据量。

通过预估的数据量获取最小处理单元的阈值，如果当前分片大小大于最小处理单元的阈值，就继续切分集合。每个分片将会生成一个

Sink 链表，当所有的分片操作完成后，ForkJoin 框架将会合并分片任何结果集。

Stream 的性能

常规数据迭代

Ø 100 的性能对比常规的迭代 > Stream 并行迭代> Stream 串行迭代

为什么这样：

1、常规迭代代码简单，越简单的代码执行效率越高。

2、Stream 串行迭代，使用了复杂的设计，导致执行速度偏低。所以是性能最低的。

3、Stream 并行迭代 使用了 Fork-Join 线程池,所以效率比 Stream 串行迭代快，但是对比常规迭代还是要慢（毕竟设计和代码复杂）

大数据迭代

Ø 一亿的数组性能对比（默认线程池）为什么这样：

1、Stream 并行迭代 使用了 Fork-Join 线程池, 而线程池线程数为 cpu 的核心数（我的电脑为 12 核），大数据场景下，能够利用多线

程机制，所以效率比 Stream 串行迭代快，同时多线程机制切换带来的开销相对来说还不算多，所以对比常规迭代还是要快（虽然设计

和代码复杂）

2、常规迭代代码简单，越简单的代码执行效率越高。

3、Stream 串行迭代，使用了复杂的设计，导致执行速度偏低。所以是性能最低的。

Ø 一亿的数组性能对比（线程池数量=2）为什么这样：

Stream 并行迭代 使用了 Fork-Join 线程池,大数据场景下，虽然利用多线程机制，但是线程池线程数为 2，所以多个请求争抢着执行任

务，想象对请求来说任务是被交替执行完成，所以对比常规迭代还是要慢（虽然用到了多线程技术）

Ø 一亿的数组性能对比（线程池数量=240）为什么这样：

Stream 并行迭代使用了 Fork-Join 线程池, 而线程池线程数为 240，大数据场景下，虽然利用多线程机制，但是线程太多，线程的上下

文切换成本过高，所以导致了执行效率反而没有常规迭代快。

如何合理使用 Stream？

我们可以看到：在循环迭代次数较少的情况下，常规的迭代方式性能反而更好；而在大数据循环迭代中， parallelStream（合理的线程池数上）有一定的优势。

但是由于所有使用并行流 parallelStream 的地方都是使用同一个 Fork-Join 线程池，而线程池线程数仅为 cpu 的核心数。

切记，如果对底层不太熟悉的话请不要乱用并行流 parallerStream（尤其是你的服务器核心数比较少的情况下）