[大数据]Java8新特性之Stream API

1 介绍

1.1 了解Stream

Java8中有两大最为重要的改变。第一个是 Lambda 表达式；另外一
个则是 Stream API(java.util.stream.*)。
Stream 是 Java8 中处理集合的关键抽象概念，它可以指定你希望对
集合进行的操作，可以执行非常复杂的查找、过滤和映射数据等操作。
使用Stream API 对集合数据进行操作，就类似于使用 SQL 执行的数
据库查询。也可以使用 Stream API 来并行执行操作。
简而言之，Stream API 提供了一种高效且易于使用的处理数据的方式。

1.2 什么是流（Stream）

是数据渠道，用于操作数据源（集合、数组等）所生成的元素序列。
“集合讲的是数据，流讲的是计算！”

注意：

• Stream 自己不会存储元素
• Stream 不会改变源对象。相反，他们会返回一个持有结果的新Stream
• Stream 操作是延迟执行的。这意味着他们会等到需要结果的时候才执行

1.3 创建步骤

• 创建 Stream
  一个数据源（如：集合、数组），获取一个流

• 中间操作
  一个数据源（如：集合、数组），获取一个流

• 终止操作(终端操作)
  一个终止操作，执行中间操作链，并产生结果

2.使用

2.1 创建 Stream

Java8 中的 Collection 接口被扩展，提供了
两个获取流的方法：

• default Stream stream() : 返回一个顺序流
• default Stream parallelStream() : 返回一个并行流

由数组创建流
	Java8 中的 Arrays 的静态方法 stream() 可以获取数组流：
		 static <T> Stream<T> stream(T[] array): 返回一个流
		  
	重载形式，能够处理对应基本类型的数组：
		public static IntStream stream(int[] array)
	 	public static LongStream stream(long[] array)
		public static DoubleStream stream(double[] array)

由值创建流
	可以使用静态方法 Stream.of(), 通过显示值
	创建一个流。它可以接收任意数量的参数
			public static<T> Stream<T> of(T... values) : 返回一个流

由函数创建流：创建无限流
	可以使用静态方法 Stream.iterate() 和Stream.generate(), 创建无限流
	*  迭代
		public static<T> Stream<T> iterate(final T seed, final  UnaryOperator<T> f) 
	*  生成
		public static<T> Stream<T> generate(Supplier<T> s) :

具体代码如下：

@Test
	public void test1(){
		// 1. Collection 提供了两个方法  stream() 与 parallelStream()
		List<String> list = new ArrayList<>();
		Stream<String> stream = list.stream(); //获取一个顺序流
		Stream<String> parallelStream = list.parallelStream(); //获取一个并行流
		
		// 2. 通过 Arrays 中的 stream() 获取一个数组流
		Integer[] nums = new Integer[10];
		Stream<Integer> stream1 = Arrays.stream(nums);
		
		// 3. 通过 Stream 类中静态方法 of()
		Stream<Integer> stream2 = Stream.of(1,2,3,4,5,6);
		
		// 4. 创建无限流
		// 迭代
		Stream<Integer> stream3 = Stream.iterate(0, (x) -> x + 2).limit(10);
		stream3.forEach(System.out::println);
		
		// 生成
		Stream<Double> stream4 = Stream.generate(Math::random).limit(2);
		stream4.forEach(System.out::println);
		
	}

2.2 中间操作

概念：
	多个中间操作可以连接起来形成一个流水线，除非流水
	线上触发终止操作，否则中间操作不会执行任何的处理！
	而在终止操作时一次性全部处理，称为“惰性求值”

2.2.1 stream之筛选与切片

筛选与切片
	filter——接收 Lambda ， 从流中排除某些元素。
	limit——截断流，使其元素不超过给定数量。
	skip(n) —— 跳过元素，返回一个扔掉了前 n 个元素的流。若流中元素不足 n 个，则返回一个空流。与 limit(n) 互补
	distinct——筛选，通过流所生成元素的 hashCode() 和 equals() 去除重复元素

List<Employee> emps = Arrays.asList(
            new Employee("李四", 59, 6666.66),
            new Employee("张三", 18, 9999.99),
            new Employee("王五", 28, 3333.33),
            new Employee("赵六", 8, 7777.77),
            new Employee("赵六", 8, 7777.77),
            new Employee("赵六", 8, 7777.77),
            new Employee("田七", 38, 5555.55)
);

// 内部迭代：迭代操作由Stream API完成
Stream<Employee> employeeStream = 
		emps.stream().filter((employee) -> employee.getAge() > 35);
		
employeeStream.forEach(System.out::println);

注意中间操作不产生任何结果
但是终止操作要执行，这个过程就叫做 “惰性求值”

//外部迭代
	@Test
	public void test3(){
		Iterator<Employee> it = emps.iterator();
		
		while(it.hasNext()){
			System.out.println(it.next());
		}
	}
	

其他操作

@Test
	public void test4(){
		emps.stream()
			.filter((e) -> {
				System.out.println("短路！"); // &&  ||
				return e.getSalary() >= 5000;
			}).limit(3)
			.forEach(System.out::println);
	}
	
	@Test
	public void test5(){
		emps.parallelStream()
			.filter((e) -> e.getSalary() >= 5000)
			.skip(2)
			.forEach(System.out::println);
	}
	
	@Test
	public void test6(){
		emps.stream()
			.distinct()
			.forEach(System.out::println);
	}

结果：

注意：distinct是通过流所生成元素的 hashCode() 和 equals() 去除重复元素
因此，所在实体类需要重写hashCode() 和 equals() 方法


重写后

2.2.2 stream之映射

映射
	map——接收 Lambda ， 将元素转换成其他形式或提取信息。
		接收一个函数作为参数，该函数会被应用到每个元素上，并将其映射成一个新的元素。
	flatMap——接收一个函数作为参数，将流中的每个值都换成另一个流，
						然后把所有流连接成一个流

应用场景一：

 @Test
   public void test() {
        List<String> list = Arrays.asList("aaa","bbb","ccc","ddd");
        list.stream().map( (str) -> {
           return str.toUpperCase();
        }).forEach(System.out::println);
    }

应用场景二：提取信息

应用场景三：

@Test
 public void test() {
    List<String> list = Arrays.asList("aaa","bbb","ccc","ddd");
    Stream<Stream<Character>> streamStream = list.stream().map(Java8_stream::filterCharacter);
    streamStream.forEach( (stream) ->stream.forEach(System.out::print) );
 }

    // 给定一个字符串，返回一个流
    public static Stream<Character> filterCharacter(String str) {
        List<Character> list = new ArrayList<>();
        for (Character character : str.toCharArray()) {
            list.add(character);
        }
        return list.stream();
    }

上面的代码使用 flatmap，这个是只形成一个流

 @Test
 public void test() {
      List<String> list = Arrays.asList("aaa","bbb","ccc","ddd");
      Stream<Character> characterStream = list.stream().flatMap(Java8_stream::filterCharacter);
      characterStream.forEach(System.out::print);
 }

应用场景四：排序

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

 @Test
    public void test7() {
        List<String> list = Arrays.asList("aaa","bbb","ccc","ddd");
        list.stream().sorted().forEach(System.out::println);

		
		emps.stream()
                .sorted((x, y) -> {
                    if(x.getAge() == y.getAge()){
                        return x.getName().compareTo(y.getName());
                    }else{
                        return Integer.compare(x.getAge(), y.getAge());
                    }
                }).forEach(System.out::println);

    }

2.3 stream的终止操作

2.3.1 查找与匹配

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

场景应用一: 匹配与查找

// 准备数据
List<Employee> emps = Arrays.asList(
            new Employee("李四", 59, 6666.66, Employee.Status.FREE),
            new Employee("张三", 18, 9999.99,Employee.Status.BUSY),
            new Employee("王五", 28, 3333.33,Employee.Status.VOCATION),
            new Employee("赵六", 8, 7777.77,Employee.Status.FREE),
            new Employee("赵六", 8, 7777.77,Employee.Status.BUSY),
            new Employee("赵六", 8, 7777.77,Employee.Status.VOCATION),
            new Employee("田七", 38, 5555.55,Employee.Status.FREE)
    );


@Test
    public void test() {
        boolean b = emps.stream().allMatch((e) -> {
            return e.getStatus().equals(Employee.Status.BUSY);
        });
        System.out.println(b);

        boolean b1 = emps.stream().anyMatch((e) -> e.getStatus().equals(Employee.Status.BUSY));
        System.out.println(b1);

        boolean b2 = emps.stream().noneMatch((e) -> e.getStatus().equals(Employee.Status.BUSY));
        System.out.println(b2);

        Optional<Employee> op = emps.stream().sorted((e1, e2) -> Double.compare(e1.getSalary(), e2.getSalary())).findFirst();
        System.out.println(op.get());

        Optional<Employee> optionalEmployee = emps.parallelStream().filter((e) -> e.getStatus().equals(Employee.Status.FREE)).findAny();
        System.out.println(optionalEmployee.get());
    }

场景应用二: 统计、最大、最小

 @Test
 public void test() {
        long count = emps.stream().count();
        System.out.println("总数："+count);

        Optional<Employee> max = emps.stream().max(Comparator.comparingDouble(Employee::getSalary));
        System.out.println("最大工资："+max.get());

        Optional<Double> min = emps.stream().map(Employee::getSalary).min(Double::compareTo);
        System.out.println("最低工资是多少："+min.get());
 }

场景应用三: 归约

	归约
		reduce(T identity, BinaryOperator)    /    reduce(BinaryOperator) 
		可以将流中元素反复结合起来，得到一个值。

备注：map 和 reduce 的连接通常称为map-reduce 模式，因 Google 用它来进行网络搜索而出名。

场景应用四: 收集

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

Collector 接口中方法的实现决定了如何对流执行收集操作(如收集到 List、Set、Map)。
但是 Collectors 实用类提供了很多静态方法，可以方便地创建常见收集器实例

@Test
    public void test() {
        // 将当前公司所有人名字提取出来，添加到集合当中去
        List<String> stringList = emps.stream().map(Employee::getName).collect(Collectors.toList());
        stringList.forEach(System.out::println);
    }

有很多子接口

		Set<String> set = emps.stream()
			.map(Employee::getName)
			.collect(Collectors.toSet());
		
		set.forEach(System.out::println);

		System.out.println("----------------------------------");
		
		HashSet<String> hs = emps.stream()
			.map(Employee::getName)
			.collect(Collectors.toCollection(HashSet::new));
		
		hs.forEach(System.out::println);

 @Test
    public void test4(){
    	// 最大值
        Optional<Double> max = emps.stream()
                .map(Employee::getSalary)
                .collect(Collectors.maxBy(Double::compare));
        System.out.println(max.get());

		// 最小值
        Optional<Employee> op = emps.stream()
                .collect(Collectors.minBy((e1, e2) -> Double.compare(e1.getSalary(), e2.getSalary())));
        System.out.println(op.get());

		// 总和
        Double sum = emps.stream()
                .collect(Collectors.summingDouble(Employee::getSalary));
        System.out.println(sum);

		// 平均值
        Double avg = emps.stream()
                .collect(Collectors.averagingDouble(Employee::getSalary));
        System.out.println(avg);

		// 总数:
        Long count = emps.stream()
                .collect(Collectors.counting());
        System.out.println(count);

        System.out.println("--------------------------------------------");

        DoubleSummaryStatistics dss = emps.stream()
                .collect(Collectors.summarizingDouble(Employee::getSalary));

        System.out.println(dss.getMax());
    }

场景应用五: 收集之分组

 // 分组
    @Test
    public void test6() {
        Map<Employee.Status, List<Employee>> collect = emps.stream().collect(Collectors.groupingBy(Employee::getStatus));
        System.out.println(collect);
    }

多级分组

  //多级分组
    @Test
    public void test6(){
        Map<Employee.Status, Map<String, List<Employee>>> map = emps.stream()
                .collect(Collectors.groupingBy(Employee::getStatus, Collectors.groupingBy((e) -> {
                    if(e.getAge() >= 60)
                        return "老年";
                    else if(e.getAge() >= 35)
                        return "中年";
                    else
                        return "成年";
                })));

        System.out.println(map);
    }

分区

    @Test
    public void test7(){
        Map<Boolean, List<Employee>> map = emps.stream()
                .collect(Collectors.partitioningBy((e) -> e.getSalary() >= 5000));

        System.out.println(map);
    }