[Java知识库]Java8新特性----Stream

What is Stream ?

注意:

Stream操作三部曲

使用演示:

    /*
    * Stream的三个操作步骤
    *
    * 1.创建stream
    * 2.中间操作
    * 3.终止操作(终端操作)
    *
    * */
    @Test
    void test()
    {
      //1.创建stream
   //(1):可以通过collection系列集合提供的stream()或者parallelStream()
        List<String> list=new ArrayList<>();
        Stream<String> stream = list.stream();
   //(2): 通过Arrays里面的静态方法stream()获取数据流
        People[] peoples=new People[10];
        Stream<People> stream1 = Arrays.stream(peoples);
    //(3):通过stream里面的静态方法of()
        Stream<String> aa = Stream.of("aa", "bb", "cc");
        //(4):创建无限流
         //迭代
      Stream.iterate(0, (x) -> x + 2)
              .limit(10)//中间操作
              .forEach(System.out::println);
    }
}

中间操作

筛选与切片

filter---接收Lambda，从流中排除某些元素

limit(max)---截断流，使其元素不超过给定数量

skip(n)---跳过元素，返回一个扔掉了前n个的元素的流，若流张元素不足n个，则返回一个空流，与limit(n)互补

distinct---筛选，通过流所生成的元素的hashcode()和equals()去重复元素

内部迭代: 迭代操作由Stream API完成

终止操作:一次性执行全部内容，即惰性求值

使用演示:

public class TestMain
{
    List<People> peopleList= Arrays.asList(
            new People("1号",18,3000),
            new People("2号",21,4000),
            new People("3号",19,5000),
            new People("4号",20,3500)
    );

    @Test
    void test()
    {
      //中间操作不会执行任何操作
        Stream<People> s=peopleList.stream().filter(people ->people.getAge()>19);
        //终止操作:一次性执行全部内容，即惰性求值
        s.forEach(System.out::println);
    }
}

外部迭代

      //外部迭代
        Iterator<People> iterator = peopleList.iterator();
        while(iterator.hasNext())
            System.out.println(iterator.next());

limit ===> 短路

public class TestMain
{
    List<People> peopleList= Arrays.asList(
            new People("1号",18,3000),
            new People("2号",21,4000),
            new People("3号",19,5000),
            new People("4号",20,3500)
    );

    @Test
    void test()
    {
    //当查询到满足条件的两条数据后，就停止迭代，此行为称为短路
peopleList.stream().
        filter(people ->{
            System.out.println("短路");
            return people.getAge()>15;}).
        limit(2).//短路
        forEach(System.out::println);
    }
}

skip ===>跳过前n个元素

public class TestMain
{
    List<People> peopleList= Arrays.asList(
            new People("1号",18,3000),
            new People("2号",21,4000),
            new People("3号",19,5000),
            new People("4号",20,3500)
    );

    @Test
    void test()
    {
peopleList.stream().
        filter(people ->{
            System.out.println("短路");
            return people.getAge()>15;}).
        skip(2).
        forEach(System.out::println);
    }
}

distinct进行元素去重(自定义类需要重写对应的hashcode和equals方法)

public class TestMain
{
    List<People> peopleList= Arrays.asList(
            new People("1号",18,3000),
            new People("2号",21,4000),
            new People("2号",21,4000),
            new People("4号",20,3500)
    );

    @Test
    void test()
    {
peopleList.stream().
        distinct().
        forEach(System.out::println);
    }
}

映射

map–接收Lambda,将元素转换为其他形式或提取信息，接收一个函数作为参数，该函数会被应用到每个元素上，并将其映射成一个新的元素

flatMap—接收一个函数作为参数，将流中的每个值都换成另一个流，然后把所有流连接成一个流

map的使用演示:

public class TestMain
{
    List<People> peopleList= Arrays.asList(
            new People("1号",18,3000),
            new People("2号",21,4000),
            new People("2号",21,4000),
            new People("4号",20,3500)
    );

    @Test
    void test()
    {
     List<String> list=Arrays.asList("a","b","c");
        //将原先集合里面的小写，全部转换为大写，并输出
     list.stream().map((x)->x.toUpperCase()).forEach(System.out::println);
     //对原先的流是没有影响的
        System.out.println(list);
        System.out.println("------------------------------------------------");
        // peopleList.stream().map(p->p.getName());
        //将原先集合里面的People元素全部转换为String元素
        peopleList.stream().map(People::getName).forEach(System.out::println);
    }
}

flatMap使用演示:

使用前，先看一下下面这个案例:

    void test()
    {
     List<String> list=Arrays.asList("aaa","bbb","ccc");
        Stream<Stream<Character>> sm = list.stream().map(TestMain::getAll);
        //相当于当前sm大流里面存放了三个小流
        sm.forEach(System.out::println);
    }

    public static Stream<Character> getAll(String str)
    {
          List<Character> list=new ArrayList<>();
          for(Character ch:str.toCharArray())
          {
              list.add(ch);
          }
          return list.stream();
    }

显然这里我们将list集合对应的新流中每一个元素，都映射为了一个流，并返回，相当于现在的大流中有三个小流

下面我们需要遍历这些小流，取出里面的值

    void test()
    {
     List<String> list=Arrays.asList("aaa","bbb","ccc");
        Stream<Stream<Character>> sm = list.stream().map(TestMain::getAll);
        //遍历大流的同时，遍历小流，取出小流中的值
        sm.forEach(x-> x.forEach(System.out::println));//效果{{a,a,a},{b,b,b},{c,c,c}}
    }

    public static Stream<Character> getAll(String str)
    {
          List<Character> list=new ArrayList<>();
          for(Character ch:str.toCharArray())
          {
              list.add(ch);
          }
          return list.stream();
    }

显然上面写法比较复杂，下面给出简化写法

    @Test
    void test()
    {
     List<String> list=Arrays.asList("aaa","bbb","ccc");
     //返回值不在是大流嵌套小流，而是一个流
        Stream<Character> characterStream = list.stream()
                .flatMap(TestMain::getAll);// 效果{a,a,a,b,b,b,c,c,c}
        characterStream.forEach(System.out::println);
    }

    public static Stream<Character> getAll(String str)
    {
          List<Character> list=new ArrayList<>();
          for(Character ch:str.toCharArray())
          {
              list.add(ch);
          }
          return list.stream();
    }

map与flatmap的区别

map是将对应的每个小流放入当前大流中构成一个流

flatmap取出集合中的每个元素放入当前的流中，相当于将每个小流里面的元素拿出来组合为一个大流

这里还可以参考add()和addAll()的关系:

     List<String> list=Arrays.asList("aaa","bbb","ccc");
   List list1=new ArrayList();
  list1.add(list);
  list1.addAll(list);
        System.out.println(list1);

排序

sorted()—自然排序(Comparable)

sorted(Comparator com)—定制排序(Comparator)

    List<People> peopleList= Arrays.asList(
            new People("1号",18,3000),
            new People("2号",21,4000),
            new People("2号",21,4000),
            new People("4号",18,3500)
    );
    @Test
    void test()
    {
        //这里people没有实现Comparable接口，因此没有自然排序的功能
        //我们需要定制排序
        peopleList.stream().sorted((x,y)->{
            if(x.getAge()==y.getAge())
                //money按照降序排列
                return -x.money.compareTo(y.getMoney());
            else
                return x.getAge().compareTo(y.getAge());
        }).forEach(System.out::println);
    }

Stream的终止操作如下

查找与匹配

查找与匹配
allMatch--检查是否匹配所有元素
anyMatch---检查是否至少匹配一个元素
noneMatch---检查是否没有匹配所有元素
findFirst---返回第一个元素
findAny---返回当前流中任意元素
count---返回流中元素的总个数
max----返回流中最大值
min---返回流中最小值

演示:

public class TestMain
{
    List<People> peopleList= Arrays.asList(
        new People("1号",18,3000, People.STATUS.BUSY),
        new People("2号",21,4000, People.STATUS.FREE),
        new People("2号",21,4000, People.STATUS.BUSY),
        new People("4号",18,3500, People.STATUS.BUSY)
);
    /*
allMatch--检查是否匹配所有元素
anyMatch---检查是否至少匹配一个元素
noneMatch---检查是否没有匹配所有元素
findFirst---返回第一个元素
findAny---返回当前流中任意元素
count---返回流中元素的总个数
max----返回流中最大值
min---返回流中最小值*/
    @Test
    void test()
    {
        boolean ret = peopleList.stream().allMatch(x -> x.getStatus().equals(People.STATUS.BUSY));
        System.out.println(ret);
        boolean ret1 = peopleList.stream().anyMatch(x -> x.getStatus().equals(People.STATUS.FREE));
        System.out.println(ret1);
        boolean ret2 = peopleList.stream().noneMatch(x -> x.getStatus().equals(People.STATUS.BUSY));
        System.out.println(ret2);
        //得到第一个元素
        Optional<People> first = peopleList.stream().sorted((x, y) -> -x.getMoney().compareTo(y.getMoney())).findFirst();
        System.out.println(first.get());
        //得到当前流中的任意一个元素
        //parallelStream:多线程并行查找
        Optional<People> any = peopleList.parallelStream().filter(x -> x.getStatus().equals(People.STATUS.FREE)).findAny();
        System.out.println(any.get());
        //元素总个数
        long count = peopleList.stream().count();
        System.out.println(count);
        //返回流中最大值和最小值
        Optional<People> max = peopleList.stream().max((x, y) -> -x.getMoney().compareTo(y.getMoney()));
        System.out.println(max.get());
        //获取当前最小的金钱数
        Optional<Integer> min = peopleList.stream().map(People::getMoney).min(Integer::compareTo);
        System.out.println(min.get());
    }
}

归约–reduce

public class TestMain
{
    List<People> peopleList= Arrays.asList(
        new People("1号",18,3000, People.STATUS.BUSY),
        new People("2号",21,4000, People.STATUS.FREE),
        new People("2号",21,4000, People.STATUS.BUSY),
        new People("4号",18,3500, People.STATUS.BUSY)
);

    @Test
    void test()
    {
//    T reduce(T identity, BinaryOperator<T> accumulator);
//这里可以使用ClassName::MethodName的原因是 Function<T, R>，比getMoeny多一个参数，且第一个参数类型为People
        Integer sum = peopleList.stream().map(People::getMoney).reduce(0, (x, y) -> x + y);//0是起始累加值
        System.out.println("money总和为:"+sum);
    }
}

也可以不指定起始值，但是这样可能数据为空，因此会被封装为一个Optional对象

public class TestMain
{
    List<People> peopleList= Arrays.asList(
        new People("1号",18,3000, People.STATUS.BUSY),
        new People("2号",21,4000, People.STATUS.FREE),
        new People("2号",21,4000, People.STATUS.BUSY),
        new People("4号",18,3500, People.STATUS.BUSY)
);

    @Test
    void test()
    {
//    T reduce(T identity, BinaryOperator<T> accumulator);
        Optional<Integer> reduce = peopleList.stream().map(People::getMoney).reduce((x, y) -> x + y);
        System.out.println("money总和为:"+reduce.get());
    }
}

这里不一定非要是数的累加，也可以是字符串的反复拼接

.reduce("",String::contact);

收集

collect----将流转换为其他形式，接收一个Collector接口的实现，用于给Stream中元素做汇总的方法

演示:

public class TestMain
{
    List<People> peopleList= Arrays.asList(
        new People("aaa",18,3000, People.STATUS.BUSY),
        new People("bbb",21,4000, People.STATUS.FREE),
        new People("ccc",21,4000, People.STATUS.BUSY),
        new People("ddd",18,3500, People.STATUS.BUSY)
);

    @Test
    void test()
    {
        //结果收集到map中
        Map<String, String> collect = peopleList.stream().map(People::getName)
        //指定key和value,这里的key是name字符串转大写，value就是name字符串本身不变
                .collect(Collectors.toMap(x -> x.toUpperCase(), y -> y));
        System.out.println(collect);
        System.out.println("==============================");
        //结果收集到List中
        List<String> stringList = peopleList.stream().map(People::getName).collect(Collectors.toList());
        System.out.println(stringList);
        System.out.println("==============================");
        //结果收集到HashSet中
        HashSet<String> collect1 = peopleList.stream().
                map(People::getName).collect(Collectors.toCollection(HashSet::new));
        System.out.println(collect1);
    }
}

collect的其他一些用法

public class TestMain
{
    List<People> peopleList= Arrays.asList(
        new People("aaa",18,3000, People.STATUS.BUSY),
        new People("bbb",21,4000, People.STATUS.FREE),
        new People("ccc",21,4000, People.STATUS.BUSY),
        new People("ddd",18,3500, People.STATUS.BUSY)
);

    @Test
    void test()
    {
        //计算当前流中元素的总数
        Long sum = peopleList.stream().collect(Collectors.counting());
        System.out.println("当前流中元素的总数:"+sum);
        //计算工资平均值
        Double MoneyAvg = peopleList.stream().collect(Collectors.averagingInt(People::getMoney));
        System.out.println("工资平均值:"+MoneyAvg);
        //计算年龄的所有信息
        IntSummaryStatistics age = peopleList.stream().collect(Collectors.summarizingInt(People::getAge));
        System.out.println("年龄所有相关的信息:"+age);
        //计算年龄的总和
        Integer ageSUm = peopleList.stream().collect(Collectors.summingInt(People::getAge));
        System.out.println(ageSUm);
        //计算工资最大值
        Optional<Integer> moneyMax = peopleList.stream().map(People::getMoney).collect(Collectors.maxBy((x,y)->Integer.compare(x,y)));
        System.out.println("最高工资:"+moneyMax.get());
        //计算最低工资
        Optional<Integer> moneyMin = peopleList.stream().map(People::getMoney).collect(Collectors.minBy(Integer::compare));
        System.out.println("最低工资:"+moneyMin.get());
    }
}

collect里面的分组

单级分组:

public class TestMain
{
    List<People> peopleList= Arrays.asList(
        new People("aaa",18,3000, People.STATUS.BUSY),
        new People("bbb",21,4000, People.STATUS.FREE),
        new People("ccc",21,4000, People.STATUS.FREE),
        new People("ddd",18,3500, People.STATUS.BUSY)
      );

    @Test
    void test()
    {
    //单级分组
        Map<People.STATUS, List<People>> collect = peopleList.stream().collect(Collectors.groupingBy(People::getStatus));
        System.out.println(collect);
    }
}

多级分组:

public class TestMain
{
    List<People> peopleList= Arrays.asList(
        new People("aaa",18,3000, People.STATUS.BUSY),
        new People("bbb",21,4000, People.STATUS.FREE),
        new People("ccc",21,10000, People.STATUS.FREE),
        new People("ddd",18,12000, People.STATUS.BUSY)
      );

    @Test
    void test()
    {
//先按照状态分组，再按照money分组
        Map<People.STATUS, Map<String, List<People>>> collect = peopleList.stream().collect(Collectors.groupingBy(People::getStatus, Collectors.groupingBy(
                x -> {
                    if (x.getMoney() >= 10000)
                        return "有钱人";
                    else
                        return "穷人";
                }
        )));
        System.out.println(collect);
    }
}

collect里面的分区

public class TestMain
{
    List<People> peopleList= Arrays.asList(
        new People("aaa",18,3000, People.STATUS.BUSY),
        new People("bbb",21,4000, People.STATUS.FREE),
        new People("ccc",21,10000, People.STATUS.FREE),
        new People("ddd",18,12000, People.STATUS.BUSY)
      );

    @Test
    void test()
    {
     //按照true or false进行分区
        Map<Boolean, List<People>> ret = peopleList.stream().collect(Collectors.partitioningBy(x -> x.getMoney() >= 10000));
        System.out.println(ret);
    }
}

collect里面获取某个属性相关的详细信息(平均值，最大值…)

public class TestMain
{
    List<People> peopleList= Arrays.asList(
        new People("aaa",18,3000, People.STATUS.BUSY),
        new People("bbb",21,4000, People.STATUS.FREE),
        new People("ccc",21,10000, People.STATUS.FREE),
        new People("ddd",18,12000, People.STATUS.BUSY)
      );

    @Test
    void test()
    {
        IntSummaryStatistics collect = peopleList.stream().collect(Collectors.summarizingInt(People::getMoney));
        System.out.println(collect);
        System.out.println(collect.getMax());
        System.out.println(collect.getCount());
    }
}

collect里面的join，完成字符串连接工作

public class TestMain
{
    List<People> peopleList= Arrays.asList(
        new People("aaa",18,3000, People.STATUS.BUSY),
        new People("bbb",21,4000, People.STATUS.FREE),
        new People("ccc",21,10000, People.STATUS.FREE),
        new People("ddd",18,12000, People.STATUS.BUSY)
      );

    @Test
    void test()
    {
        //第一个参数是连接字符串时分割的符合，后面两个参数依次是前缀和后缀
        String ret = peopleList.stream().map(People::getName).collect(Collectors.joining(",", "==", "=="));
        System.out.println(ret);
    }
}

并行流与串行流

一、什么是并行流

并行流 : 就是把一个内容分成多个数据块，并用不同的线程分 别处理每个数据块的流。

Java 8 中将并行进行了优化，我们可以很容易的对数据进行并 行操作。Stream API 可以声明性地通过 parallel() 与 sequential() 在并行流与顺序流之间进行切换。

二、了解 Fork/Join框架

Fork/Join 框架 : 就是在必要的情况下，将一个大任务，进行拆分(fork)成若干个 小任务(拆到不可再拆时)，再将一个个的小任务运算的结果进行 join 汇总.

三、Fork/Join 框架与传统线程池的区别

采用 “工作窃取”模式(work-stealing)： 当执行新的任务时它可以将其拆分分成更小的任务执行，并将小任务加到线 程队列中，然后再从一个随机线程的队列中偷一个并把它放在自己的队列中。

相对于一般的线程池实现，fork/join框架的优势体现在对其中包含的任务的 处理方式上.在一般的线程池中，如果一个线程正在执行的任务由于某些原因 无法继续运行,那么该线程会处于等待状态。而在fork/join框架实现中，如果 某个子问题由于等待另外一个子问题的完成而无法继续运行.那么处理该子 问题的线程会主动寻找其他尚未运行的子问题来执行。这种方式减少了线程 的等待时间， 高了性能。

四、 案例

创建一个ForkJoinCalculate计算类：

public class ForkJoinCalculate extends RecursiveTask<Long> {

   private long start;
   private long end;

   private static final long THRESHOLD = 1000000;

   public ForkJoinCalculate(long start, long end) {
       this.start = start;
       this.end = end;
   }


   @Override
   protected Long compute() {
       long length = end - start;

       if (length <= THRESHOLD) {
           long sum = 0;

           for (long i = start; i <= end; i++) {
               sum += i;
           }
           return sum;
       }else {
           long middle = (start + end) / 2;
           ForkJoinCalculate left = new ForkJoinCalculate(start, middle);
           left.fork();

           ForkJoinCalculate right = new ForkJoinCalculate(middle + 1, end);
           right.fork();

           return left.join() + right.join();
       }
   }
}

测试方法：

private static final long END_VALUE = 10000000000L;

// fork join
@Test
public void test1(){
   Instant start = Instant.now();

   ForkJoinPool pool = new ForkJoinPool();
   ForkJoinTask<Long> task = new ForkJoinCalculate(0, END_VALUE);

   Long sum = pool.invoke(task);
   System.out.println(sum);

   Instant end = Instant.now();
   System.out.println("耗时：" + Duration.between(start, end).toMillis());
}

执行结果：

-5340232216128654848
耗时：2325

使用普通for 循环：

@Test
public void test2(){
    Instant start = Instant.now();

    long sum = 0L;

    for (long i = 0; i <= END_VALUE; i ++){
        sum += i;
    }

    System.out.println(sum);

    Instant end = Instant.now();

    System.out.println("耗时：" + Duration.between(start, end).toMillis());
}

执行结果：

-5340232216128654848
耗时：3571

java8中 Fork/Join计算

//java8 的并行流测试
@Test
public  void test3(){
    Instant start = Instant.now();

    LongStream.rangeClosed(0, END_VALUE)
            .parallel()
            .reduce(0, Long::sum);

    Instant end = Instant.now();

    System.out.println("耗时为：" + Duration.between(start, end).toMillis());
}

执行结果：

耗时为：1690

检查本机的可用处理器数：

 // 可用处理器
 @Test
 public  void test4(){
     int num = Runtime.getRuntime().availableProcessors();
     System.out.println(num);
 }

执行结果：

8

Optional类

Java 8 Optional的正确姿势