Java8新特性中篇(Stream流)

了解Stream

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

什么是Stream

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

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

Stream的三个操作步骤

1.创建Stream
2.中间操作
3.终止操作（终端操作）

创建Stream

Java8中的Collection接口被扩展，提供了两个获取流的方法：
default Stream Stream():返回一个顺序流
default Stream parallelStream():返回一个并行流

package StreamAPI;

import org.junit.Test;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.regex.Matcher;
import java.util.stream.Stream;

/*
 * 一、Stream的三个操作步骤
 * 1.创建Stream
 * 2.中间操作
 * 3.终止操作（终端操作）
 */
public class TestStreamAPI1 {
    //创建Stream
    @Test
    public void test1(){
        //可以通过Collection系列集合提供的stream()或parallelStream()
        List<String> list = new ArrayList<>();
        Stream<String> stream = list.stream();

        //可以通过Arrays中的静态方法stream()获取数组流
        Stream<String> stream1 = Arrays.stream(new String[10]);

        //通过Stream类中的静态方法of()
        Stream<String> stream2 = Stream.of("wo","ai","chi","kao","ji");

        //创建无限流
        //迭代
        Stream<Integer> stream3 = Stream.iterate(0,(x) -> x+5);
        stream3.limit(10).forEach(System.out::println);

        //生成
        Stream<Double> stream4 = Stream.generate(() -> Math.random());
        stream4.limit(10).forEach(System.out::println);
    }
}

Stream的中间操作

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

筛选与切片

映射

排序

示例

package StreamAPI;

import java.util.Objects;

enum Status{
    FREE,
    BUSY,
    VOCATION;
}

public class Employee {
    private String name;
    private int age;
    private double salary;
    private Status status;

    public Employee() {
    }

    public Employee(String name, int age, double salary, Status status) {
        this.name = name;
        this.age = age;
        this.salary = salary;
        this.status = status;
    }

    public String getName() {
        return name;
    }

    public void setName(String name) {
        this.name = name;
    }

    public int getAge() {
        return age;
    }

    public void setAge(int age) {
        this.age = age;
    }

    public double getSalary() {
        return salary;
    }

    public void setSalary(double salary) {
        this.salary = salary;
    }

    public Status getStatus() {
        return status;
    }

    public void setStatus(Status status) {
        this.status = status;
    }

    @Override
    public String toString() {
        return "Employee{" +
                "name='" + name + '\'' +
                ", age=" + age +
                ", salary=" + salary +
                ", status=" + status +
                '}';
    }

    @Override
    public boolean equals(Object o) {
        if (this == o) return true;
        if (o == null || getClass() != o.getClass()) return false;
        Employee employee = (Employee) o;
        return age == employee.age &&
                Double.compare(employee.salary, salary) == 0 &&
                Objects.equals(name, employee.name) &&
                status == employee.status;
    }

    @Override
    public int hashCode() {
        return Objects.hash(name, age, salary, status);
    }
}

package StreamAPI;

import LambdaExpre.Employee;
import org.junit.Test;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Iterator;
import java.util.List;
import java.util.stream.Stream;

/*
 * 一、Stream的三个操作步骤
 * 1.创建Stream
 * 2.中间操作
 * 3.终止操作（终端操作）
 */
public class TestStreamAPI2 {
    List<Employee> employees = Arrays.asList(
            new Employee("张三",18,10000),
            new Employee("赵六",30,5000),
            new Employee("李四",24,15000),
            new Employee("王五",14,8000),
            new Employee("田七",29,7000)
    );
    //中间操作

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

    //内部迭代：迭代操作由Stream API 完成
    @Test
    public void test1(){
        //中间操作:不会执行任何操作
        Stream<Employee> stream = employees.stream()
                .filter((e) -> {
                    System.out.println("Stream API的中间操作");
                    return e.getAge() > 25;
                });
        //终止操作：一次性执行全部内容，即“惰性求值”
        stream.forEach(System.out::println);
    }

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

    @Test
    public void test3(){
        employees.stream()
                .filter((e)->{
                    System.out.println("短路");
                    return e.getSalary()>6000;
                })
                .limit(2)//limit找到两个就不继续找了，短路，与&&、||类似
                .forEach(System.out::println);
    }

    @Test
    public void test4(){
        employees.stream()
                .filter((e)->e.getSalary()>6000)
                .skip(2)
                .forEach(System.out::println);
    }

    @Test
    public void test5(){
        employees.stream()
                .filter((e)->e.getSalary()>6000)
                .skip(2)
                .distinct()
                .forEach(System.out::println);
    }

    /*
       映射
       map -- 接收Lambda，将元素转换成其他形式或提取信息。接收一个函数作为参数，该函数会被应用到每个元素上，并将其映射成一个新元素。
       flatMap -- 接收一个函数作为参数，将流中的每个值都换成另一个流，然后把所有流连接成一个流
     */
    @Test
    public void test6(){
        List<String> list = Arrays.asList("wo","ai","chi","da","zha","xie");
        list.stream()
                .map((s)->s.toUpperCase())
                .forEach(System.out::println);
        System.out.println("===========================================");
        employees.stream()
                //.map((e)->e.getName())
                .map(Employee::getName)
                .forEach(System.out::println);
        System.out.println("===========================================");
        Stream<Stream<Character>> stream = list.stream()
                .map(TestStreamAPI2::filterCharacter);
        stream.forEach((x)->x.forEach(System.out::println));
        System.out.println("===========================================");
        list.stream()
                .flatMap(TestStreamAPI2::filterCharacter)
                .forEach(System.out::println);
    }
    public static Stream<Character> filterCharacter(String str){
        List<Character> list = new ArrayList<>();
        for (char c : str.toCharArray()) {
            list.add(c);
        }
        return list.stream();
    }

    @Test
    public void test7(){
        //map和flatMap的不同类似于集合的函数add(Object obj)和addAll(Collection coll)
        List<String> list = Arrays.asList("wen","yu","er");
        List list1 = new ArrayList();
        list1.add("ping");
        list1.add("fan");
        list1.add(list);
        System.out.println(list1);

        List list2 = new ArrayList();
        list2.add("ping");
        list2.add("fan");
        list2.addAll(list);
        System.out.println(list2);
    }

    /*
       排序
       sorted() -- 自然排序（Comparable）
       sorted(Comparator com) -- 定制排序（Comparator）
     */
    @Test
    public void test8(){
        List<String> list = Arrays.asList("dd","cc","nn","aa","pp","bb");
        list.stream()
                .sorted()
                .forEach(System.out::println);
        employees.stream()
                .sorted((e1,e2)->{
                    Integer num = Integer.compare(e1.getAge(),e2.getAge());
                    if(num==0){
                        return e1.getName().compareTo(e2.getName());
                    }
                    return num;
                })
                .forEach(System.out::println);
    }
}

Stream的终止操作

终端操作会从流的流水线生成结果。其结果可以是任何不是流的值，例如：List、Integer,甚至是void。

查找与匹配

归约

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

收集

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

示例

package StreamAPI;

import org.junit.Test;

import javax.swing.plaf.nimbus.State;
import java.util.*;
import java.util.stream.Collectors;

/*
 * 一、Stream的三个操作步骤
 * 1.创建Stream
 * 2.中间操作
 * 3.终止操作（终端操作）
 */
public class TestStreamAPI3 {
    //终止操作
    List<Employee> employees = Arrays.asList(
            new Employee("张三",60,10000, Status.FREE),
            new Employee("赵六",35,5000,Status.BUSY),
            new Employee("李四",24,15000, Status.VOCATION),
            new Employee("王五",20,8000,Status.FREE),
            new Employee("田七",31,7000,Status.BUSY)
    );

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

    @Test
    public void test(){
        boolean b = employees.stream()
                .allMatch((e) -> e.getStatus().equals(Status.BUSY));
        System.out.println(b);
        System.out.println("=========================================");
        boolean b1 = employees.stream()
                .anyMatch((e) -> e.getStatus().equals(Status.BUSY));
        System.out.println(b1);
        System.out.println("=========================================");
        boolean b2 = employees.stream()
                .noneMatch((e) -> e.getStatus().equals(Status.BUSY));
        System.out.println(b2);
        System.out.println("=========================================");
        Optional<Employee> first = employees.stream()
                .sorted((e1, e2) -> Integer.compare(e1.getAge(), e2.getAge()))
                .findFirst();
        System.out.println(first.get());
        System.out.println("=========================================");
        Optional<Employee> any = employees.parallelStream()
                .filter((e) -> e.getStatus().equals(Status.BUSY))
                .findAny();
        System.out.println(any.get());
        System.out.println("=========================================");
    }
    @Test
    public void test1(){
        long count = employees.stream()
                .count();
        System.out.println(count);

        Optional<Employee> max = employees.stream()
                .max((e1, e2) -> Double.compare(e1.getSalary(), e2.getSalary()));
        System.out.println(max.get());
        System.out.println("================================================");
        Optional<Double> min = employees.stream()
                .map(Employee::getSalary)
                .min(Double::compare);
        System.out.println(min.get());
        System.out.println("================================================");
    }

    /*
        归约
        reduce(T identity,BinaryOperator)--可以将流中元素反复结合起来，得到一个值。返回T
        reduce(BinaryOprator)--可以将流中元素反复结合起来，得到一个值。返回Optional<T>
        备注：map和reduce的连接通常称为map-reduce模式，因Google用它来进行网络搜索而出名。
     */
    @Test
    public void test2(){
        List<Integer> list = Arrays.asList(1,2,3,4,5,6,7,8,9,10);
        Integer reduce = list.stream()
                .reduce(0, (x, y) -> x + y);
        System.out.println(reduce);
        System.out.println("===================================");
        Optional<Double> reduce1 = employees.stream()
                .map(Employee::getSalary)
                .reduce(Double::sum);
        System.out.println(reduce1.get());
    }

    /*
       收集
       collect--将流转换为其他形式。接收一个Collector接口的实现，用于给Stream中元素做汇总的方法
     */
    @Test
    public void test3(){
        List<String> collect = employees.stream()
                .map(Employee::getName)
                .collect(Collectors.toList());
        System.out.println(collect);
        System.out.println("=====================================");
        Set<String> collect1 = employees.stream()
                .map(Employee::getName)
                .collect(Collectors.toSet());
        System.out.println(collect1);
        System.out.println("=====================================");
        HashSet<String> collect2 = employees.stream()
                .map(Employee::getName)
                .collect(Collectors.toCollection(HashSet::new));
        System.out.println(collect2);
        System.out.println("=====================================");
    }
    @Test
    public void test4(){
        //总数
        Long count = employees.stream()
                .collect(Collectors.counting());
        System.out.println(count);
        System.out.println("=====================================");
        //平均值
        Double average = employees.stream()
                .collect(Collectors.averagingDouble(Employee::getSalary));
        System.out.println(average);
        System.out.println("=====================================");
        //总和
        Double sum = employees.stream()
                .collect(Collectors.summingDouble(Employee::getSalary));
        System.out.println(sum);
        System.out.println("=====================================");
        //最大值
        Optional<Double> max = employees.stream()
                .map(Employee::getSalary)
                .collect(Collectors.maxBy(Double::compare));
        System.out.println(max.get());
        System.out.println("=====================================");
        //最小值
        Optional<Employee> min = employees.stream()
                .collect(Collectors.minBy((e1, e2) -> {
                    int num = (int) Double.compare(e1.getSalary(), e2.getSalary());
                    if (num == 0) {
                        return e1.getName().compareTo(e2.getName());
                    }
                    return num;
                }));
        System.out.println(min.get());
    }
    //分组
    @Test
    public void test5(){
        Map<Status, List<Employee>> collect = employees.stream()
                .collect(Collectors.groupingBy(Employee::getStatus));
        System.out.println(collect);
    }
    //多级分组
    @Test
    public void test6(){
        Map<Status, Map<String, List<Employee>>> collect = employees.stream()
                .collect(Collectors.groupingBy(Employee::getStatus, Collectors.groupingBy((e) -> {
                    if (e.getAge() <= 30) {
                        return "青年";
                    } else if (e.getAge() <= 50) {
                        return "中年";
                    } else {
                        return "老年";
                    }
                })));
        System.out.println(collect);
    }
    //分区
    @Test
    public void test7(){
        Map<Boolean, List<Employee>> collect = employees.stream()
                .collect(Collectors.partitioningBy((e) -> e.getSalary() >= 8000));
        System.out.println(collect);
    }

    @Test
    public void test8(){
        DoubleSummaryStatistics collect = employees.stream()
                .collect(Collectors.summarizingDouble(Employee::getSalary));
        System.out.println(collect.getCount());
        System.out.println(collect.getAverage());
        System.out.println(collect.getSum());
        System.out.println(collect.getMax());
        System.out.println(collect.getMin());
    }

    @Test
    public void test9(){
        String collect = employees.stream()
                .map(Employee::getName)
                .collect(Collectors.joining(",","=","="));
        System.out.println(collect);
    }

}

练习

1、给定一个数字列表，如何返回一个由每个数的平方构成的列表呢？
给定【1,2,3,4,5】，应该返回【1,4,9,16,25】

package StreamAPI;

import org.junit.Test;

import java.util.Arrays;

//练习
public class TestStreamAPI4 {
    /*
        给定一个数字列表，如何返回一个由每个数的平方构成的列表呢？
        给定【1,2,3,4,5】，应该返回【1,4,9,16,25】
     */
    @Test
    public void test(){
        Integer[] num = {1,2,3,4,5};
        Arrays.stream(num)
                .map((x)->x*x)
                .forEach(System.out::println);
    }
}

2、

package StreamAPI;

import java.util.Objects;

public class Trader {
    private String name;
    private String city;

    public Trader() {
    }

    public Trader(String name, String city) {
        this.name = name;
        this.city = city;
    }

    public String getName() {
        return name;
    }

    public void setName(String name) {
        this.name = name;
    }

    public String getCity() {
        return city;
    }

    public void setCity(String city) {
        this.city = city;
    }

    @Override
    public String toString() {
        return "Trader{" +
                "name='" + name + '\'' +
                ", city='" + city + '\'' +
                '}';
    }

    @Override
    public boolean equals(Object o) {
        if (this == o) return true;
        if (o == null || getClass() != o.getClass()) return false;
        Trader trader = (Trader) o;
        return Objects.equals(name, trader.name) &&
                Objects.equals(city, trader.city);
    }

    @Override
    public int hashCode() {
        return Objects.hash(name, city);
    }
}

package StreamAPI;

import java.util.Objects;

public class Transaction {
    private Trader trader;
    private Integer year;
    private  Integer value;

    public Transaction() {
    }

    public Transaction(Trader trader, Integer year, Integer value) {
        this.trader = trader;
        this.year = year;
        this.value = value;
    }

    public Trader getTrader() {
        return trader;
    }

    public void setTrader(Trader trader) {
        this.trader = trader;
    }

    public Integer getYear() {
        return year;
    }

    public void setYear(Integer year) {
        this.year = year;
    }

    public Integer getValue() {
        return value;
    }

    public void setValue(Integer value) {
        this.value = value;
    }

    @Override
    public String toString() {
        return "Transaction{" +
                "trader=" + trader +
                ", year=" + year +
                ", value=" + value +
                '}';
    }

    @Override
    public boolean equals(Object o) {
        if (this == o) return true;
        if (o == null || getClass() != o.getClass()) return false;
        Transaction that = (Transaction) o;
        return Objects.equals(trader, that.trader) &&
                Objects.equals(year, that.year) &&
                Objects.equals(value, that.value);
    }

    @Override
    public int hashCode() {
        return Objects.hash(trader, year, value);
    }
}

package StreamAPI;

import org.junit.Test;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Optional;
import java.util.stream.Stream;

public class TestTransaction {

    Trader raoul = new Trader("Raoul","Cambridge");
    Trader mario = new Trader("Mario","Milan");
    Trader alan = new Trader("Alan","Cambridge");
    Trader brian = new Trader("Brian","Cambridge");

    List<Transaction> transactions = Arrays.asList(
            new Transaction(brian,2011,300),
            new Transaction(raoul,2012,1000),
            new Transaction(raoul,2011, 400),
            new Transaction(mario,2012,710),
            new Transaction(alan,2012,950)
    );

    //1.找出2011年发生的所有交易，并按交易额排序
    @Test
    public void test(){
        transactions.stream()
                .filter((e)->e.getYear().equals(2011))
                .sorted((e1,e2)->Integer.compare(e1.getValue(),e2.getValue()))
                .forEach(System.out::println);
    }
    //2.交易员都在哪些不同的城市工作过？
    @Test
    public void test1(){
        transactions.stream()
                .map((e)->e.getTrader().getCity())
                .distinct()
                .forEach(System.out::println);
    }
    //3.查找所有来自剑桥的交易员，并按姓名排序
    @Test
    public void test2(){
        transactions.stream()
                .filter((e)->e.getTrader().getCity().equals("Cambridge"))
                .map(Transaction::getTrader)
                .sorted((e1,e2)->e1.getName().compareTo(e2.getName()))
                .distinct()
                .forEach(System.out::println);
    }
    //4.返回所有交易员的姓名字符串，按字母排序
    @Test
    public void test3(){
        transactions.stream()
                .map((e)->e.getTrader().getName())
                .distinct()
                .sorted((e1,e2)->e1.compareTo(e2))
                .forEach(System.out::println);
        System.out.println("====================================");
        String reduce = transactions.stream()
                .map((e) -> e.getTrader().getName())
                .distinct()
                .sorted((e1, e2) -> e1.compareTo(e2))
                .reduce("", String::concat);
        System.out.println(reduce);
        System.out.println("====================================");
        transactions.stream()
                .map((e) -> e.getTrader().getName())
                .distinct()
                .flatMap(TestTransaction::filterCharacter)
                .sorted((s1,s2)->s1.compareToIgnoreCase(s2))
                .forEach(System.out::println);
    }
    public static Stream<String> filterCharacter(String str){
        List<String> list = new ArrayList<>();
        for (Character c : str.toCharArray()) {
            list.add(String.valueOf(c));
        }
        return list.stream();
    }

    //有没有交易员是在米兰工作的？
    @Test
    public void test4(){
        boolean match = transactions.stream()
                .map(Transaction::getTrader)
                .distinct()
                .anyMatch((e) -> e.getCity().equals("Milan"));
        System.out.println(match);
    }
    //6.打印生活在剑桥的交易员的所有交易额
    @Test
    public void test5(){
        Optional<Integer> sum = transactions.stream()
                .filter((e) -> e.getTrader().getCity().equals("Cambridge"))
                .map(Transaction::getValue)
                .reduce(Integer::sum);
        System.out.println(sum.get());
    }
    //7.所有交易中，最高的交易额是多少
    @Test
    public void test6(){
        Optional<Integer> max = transactions.stream()
                .map(Transaction::getValue)
                .max(Integer::compare);
        System.out.println(max.get());
    }
    //8.找到交易额最小的交易
    @Test
    public void test7(){
        Optional<Transaction> first = transactions.stream()
                .sorted((o1, o2) -> Integer.compare(o1.getValue(), o2.getValue()))
                .findFirst();
        System.out.println(first.get());
        System.out.println("=======================================");
        Optional<Transaction> min = transactions.stream()
                .min((o1, o2) -> Integer.compare(o1.getValue(), o2.getValue()));
        System.out.println(min.get());
    }
}

并行流与顺序流

并行流就是把一个内容分成多个数据块，并用不同的线程分别处理每个数据块的流。
Java8中将并行进行了优化，我们可以很容易的对数据进行并行操作。Sream API可以声明性地通过parallel()与sequential()在并行流与顺序流之间进行切换。

package ForkJoin;

import java.util.concurrent.*;

public class ForkJoinCalculate extends RecursiveTask<Long> {

    private static final long serialVersionUID = -3551170146940096345L;

    private long start;
    private long end;
    private static final long THRESHOLD = 10000;

    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();
            Long leftvalue = left.join();
            Long rightvalue = right.join();
            return leftvalue+rightvalue;
        }
    }
}

package ForkJoin;

import org.junit.Test;

import java.time.Duration;
import java.time.Instant;
import java.util.OptionalLong;
import java.util.concurrent.*;
import java.util.stream.LongStream;

public class TestForkJoin {
    /*
       ForkJoin框架
     */
    @Test
    public void test(){
        Instant start = Instant.now();

        ForkJoinPool pool = new ForkJoinPool();
        RecursiveTask<Long> task = new ForkJoinCalculate(0,50000000000L);
        Long result = pool.invoke(task);
        System.out.println(result);

        Instant end = Instant.now();
        pool.shutdown();

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

    /*
       普通for
     */
    @Test
    public void test1(){
        Instant start = Instant.now();

        long sum = 0L;
        for(long i=0;i<=50000000000L;i++){
            sum+=i;
        }
        System.out.println(sum);
        Instant end = Instant.now();
        System.out.println("耗费时间为："+Duration.between(start,end).toMillis());
    }

    /*
       并行流
     */
    @Test
    public void test2(){
        Instant start = Instant.now();

        Long result = LongStream.rangeClosed(0, 10000000000L)
                .parallel()
                .reduce(0,Long::sum);
        System.out.println(result);

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

ForkJoin框架在jdk1.7之后就有了，但写起来太过复杂，并行流即简洁又快速。
ForkJoin框架详见我的另一篇博文