Java新特性-stream流

Java 8 API添加了一个新的抽象称为流Stream，可以让你以一种声明的方式处理数据。

Stream 使用一种类似用 SQL 语句从数据库查询数据的直观方式来提供一种对 Java 集合运算和表达的高阶抽象。

Stream API可以极大提高Java程序员的生产力，让程序员写出高效率、干净、简洁的代码。

这种风格将要处理的元素集合看作一种流， 流在管道中传输， 并且可以在管道的节点上进行处理， 比如筛选， 排序，聚合等。

元素流在管道中经过中间操作（intermediate operation）的处理，最后由最终操作(terminal operation)得到前面处理的结果。

+--------------------+       +------+   +------+   +---+   +-------+
| stream of elements +-----> |filter+-> |sorted+-> |map+-> |collect|
+--------------------+       +------+   +------+   +---+   +-------+

以上的流程转换为 Java 代码为：

List<Integer> transactionsIds = 
widgets.stream()
             .filter(b -> b.getColor() == RED)
             .sorted((x,y) -> x.getWeight() - y.getWeight())
             .mapToInt(Widget::getWeight)
             .sum();

关于中间函数的详解，下文讲

什么是 Stream？

Stream（流）是一个来自数据源的元素队列并支持聚合操作

• <strong元素队列< strong="">元素是特定类型的对象，形成一个队列。 Java中的Stream并不会存储元素，而是按需计算。
• 数据源 流的来源。 可以是集合，数组，I/O channel， 产生器generator 等。
• 聚合操作 类似SQL语句一样的操作， 比如filter, map, reduce, find, match, sorted等。

和以前的Collection操作不同， Stream操作还有两个基础的特征：

• Pipelining: 中间操作都会返回流对象本身。 这样多个操作可以串联成一个管道， 如同流式风格（fluent style）。 这样做可以对操作进行优化， 比如延迟执行(laziness)和短路( short-circuiting)。
• 内部迭代： 以前对集合遍历都是通过Iterator或者For-Each的方式, 显式的在集合外部进行迭代， 这叫做外部迭代。 Stream提供了内部迭代的方式， 通过访问者模式(Visitor)实现。

生成流的方式

在 Java 8 中, 集合接口有2种方法来生成流：

• stream() − 为集合创建串行流。

• parallelStream() − 为集合创建并行流。

List<String> strings = Arrays.asList("abc", "", "bc", "efg", "abcd","", "jkl");
List<String> filtered = strings.stream().filter(string -> !string.isEmpty()).collect(Collectors.toList());

Stream<Integer> stream3 = Stream.of(1, 2, 3);

其他的几种创建流的方式

• Stream的静态方法of：

/**
* 创建流
*/
@Test
public void test01(){
//Stream 静态方法
    //Stream.of(...)
    Stream<Integer> stream3 = Stream.of(1, 2, 3);
}

• Stream的静态方法iterate

//迭代
    Stream<Integer> stream4 = Stream.iterate(0, (x) -> x+2);
    stream4.limit(10).forEach(System.out::println);

其中iterate方法解释如下：

static <T> Stream<T> iterate(T seed, UnaryOperator<T> f) 
返回有序无限连续 Stream由函数的迭代应用产生 f至初始元素 seed ，产生 Stream包括 seed ， f(seed) ， f(f(seed)) ，等  

UnaryOperator是一个函数式接口：

• Stream的静态方法generate

    //生成
    Stream.generate(() -> Math.random())
        .limit(5)
        .forEach(System.out::println);

其中generate方法解释如下：

static <T> Stream<T> generate(Supplier<T> s) 
返回无限顺序无序流，其中每个元素由提供的 Supplier 。  

传递一个函数式接口，lamabda表达式作为参数。

其中forEach方法解释如下：

void forEach(Consumer<? super T> action) 
对此流的每个元素执行操作。  

传递一个函数式接口，lamabda表达式作为参数

Stream的中间操作函数

多个中间操作可以连接起来形成一个流水线，除非流水线上触发终止操作，否则中间操作不会执行任何的处理！

而在终止操作时一次性全部处理，称为”惰性求值“

筛选与切片

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

List<Employee> emps = Arrays.asList(
    new Employee(101, "Z3", 19, 9999.99),
    new Employee(102, "L4", 20, 7777.77),
    new Employee(103, "W5", 35, 6666.66),
    new Employee(104, "Tom", 44, 1111.11),
    new Employee(105, "Jerry", 60, 4444.44)
);

@Test
public void test01(){
    emps.stream()
        .filter((x) -> x.getAge() > 35)
        .limit(3) //短路？达到满足不再内部迭代
        .distinct()
        .skip(1)
        .forEach(System.out::println);

}

映射map

• map：接收函数式接口，将元素转换为其他形式(简单的替换)或提取信息（传入其他对象当中存在的方法）；接受一个函数作为参数，该函数会被应用到每个元素上，并将其映射成一个新的元素

<R> Stream<R> map(Function<? super T,? extends R> mapper) 
返回由给定函数应用于此流的元素的结果组成的流。 

• flatMap：接收一个函数式接口。优化遍历情形：将流中的对象都换成流（即流的嵌套），然后把所有流重新连接成一个流，返回的对象不再是嵌套流，而是流对象。便于遍历

<R> Stream<R> flatMap(Function<? super T,? extends Stream<? extends R>> mapper) 
返回由通过将提供的映射函数应用于每个元素而产生的映射流的内容来替换该流的每个元素的结果的流。  

map示例：接收一个lamabda表达式

@Test
public void test02(){
    List<String> list = Arrays.asList("a", "b", "c");
    list.stream()
        .map((str) -> str.toUpperCase())
        .forEach(System.out::println);
}

其中forEach(System.out::println)是lamabda表达式的简化写法，等效于

forEach(x -> {
            System.out.println(x);
        });

map存在的问题：假设现在存在一个自定义的流生成函数filterCharacter

class TestStreamApi2{
public Stream<Character> filterCharacter(String str){
    List<Character> list = new ArrayList<>();
    for (char c : str.toCharArray()) {
        list.add(c);
    }

    return list.stream();
}
@Test
public void test(){
    List<String> list = Arrays.asList("a", "b", "c");

    TestStreamApi2 test02 = new TestStreamApi2(); 

    Stream<Stream<Character>> stream= list.stream().Map(test02::filterCharacter)； 

    //遍历嵌套流 
    stream.forEach((sm)-> sm.forEach(System.out::println) );
 }
 }

其中Map(test02::filterCharacter)是lamabda表达式的简化写法，等效于

Map(str->{

    test02.filterCharacter(str)}
);

我们知道流处理中间操作毋庸置疑的是返回流，而自定义的filterCharacter(String str)仍然返回流，即流中嵌套流，返回最终返回的是Stream<Stream<Character>>

Stream<Stream<Character>>=

list.stream().flatMap(test02::filterCharacter)

这样的遍历多麻烦啊！！

flatMap示例：接收一个函数

public Stream<Character> filterCharacter(String str){
    List<Character> list = new ArrayList<>();
    for (char c : str.toCharArray()) {
        list.add(c);
    }

    return list.stream();
}

@Test
public void test03(){
    List<String> list = Arrays.asList("a", "b", "c");
    Test02 test02 = new Test02();
    list.stream()
        .flatMap(testStreamApi2::filterCharacter)
        .forEach(System.out::println);
}

排序

• sorted()：自然排序

Stream<T> sorted() 
返回由此流的元素组成的流，根据自然顺序排序。  

• sorted(Comparator c)：定制排序

Stream<T> sorted(Comparator<? super T> comparator) 
返回由该流的元素组成的流，根据提供的 Comparator进行排序。  

Comparable：自然排序

@Test
public void test04(){
    List<Integer> list = Arrays.asList(1,2,3,4,5);
    list.stream()
        .sorted() //comparaTo()
        .forEach(System.out::println);
}

Comparator：定制排序

@Test
public void test05(){
    emps.stream()
        .sorted((e1, e2) -> { //compara()
            if (e1.getAge().equals(e2.getAge())){
                return e1.getName().compareTo(e2.getName());
            } else {
                return e1.getAge().compareTo(e2.getAge());
            }
        })
        .forEach(System.out::println);
}

查找 / 匹配

终止操作：

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

public enum Status {
    FREE, BUSY, VOCATION;
}

@Test
public void test01(){
    List<Status> list = Arrays.asList(Status.FREE, Status.BUSY, Status.VOCATION);
    //s代指list中的遍历元素
    boolean flag1 = list.stream()
        .allMatch((s) -> s.equals(Status.BUSY));
    System.out.println(flag1);

    boolean flag2 = list.stream()
        .anyMatch((s) -> s.equals(Status.BUSY));
    System.out.println(flag2);

    boolean flag3 = list.stream()
        .noneMatch((s) -> s.equals(Status.BUSY));
    System.out.println(flag3);

    // 避免空指针异常
    Optional<Status> op1 = list.stream()
        .findFirst();
    // 如果Optional为空 找一个替代的对象
    Status s1 = op1.orElse(Status.BUSY);
    System.out.println(s1);

    Optional<Status> op2 = list.stream()
        .findAny();
    System.out.println(op2);

    long count = list.stream()
        .count();
    System.out.println(count);
}

归约 / 收集

• 归约：reduce可以将流中的数据反复结合起来，得到一个值

Optional<T> reduce(BinaryOperator<T> accumulator) 
使用 associative累积函数对此流的元素执行 reduction ，并返回描述减小值的 Optional （如果有）。  
T reduce(T identity, BinaryOperator<T> accumulator) 
使用提供的身份值和 associative累积功能对此流的元素执行 reduction ，并返回减小的值。  
<U> U reduce(U identity, BiFunction<U,? super T,U> accumulator, BinaryOperator<U> combiner) 
执行 reduction在此流中的元素，使用所提供的身份，积累和组合功能。  

• 收集：collect 将流转换成其他形式；接收一个 Collector 接口的实现，用于给流中元素做汇总的方法

<R,A> R collect(Collector<? super T,A,R> collector) 
使用 Collector对此流的元素执行 mutable reduction Collector 。  
<R> R collect(Supplier<R> supplier, BiConsumer<R,? super T> accumulator, BiConsumer<R,R> combiner) 
对此流的元素执行 mutable reduction操作。  

reduce：求和案例

/**
* Java：
*  - reduce：需提供默认值（初始值）
* Kotlin：
*  - fold：不需要默认值（初始值）
*  - reduce：需提供默认值（初始值）
*/
@Test
public void test01(){
    List<Integer> list = Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9);
    Integer sum= list.stream()
        .reduce(0, (x, y) -> x + y);
    System.out.println(sum);
}

Collect收集案例：collect(Collector c)其中Collector接口方法中的实现决定了如何对流执行收集操作（如收集到List，Set，Map）。

<R,A> R collect(Collector<? super T,A,R> collector) 
使用 Collector对此流的元素执行 mutable reduction Collector 。  
<R> R collect(Supplier<R> supplier, BiConsumer<R,? super T> accumulator, BiConsumer<R,R> combiner) 
对此流的元素执行 mutable reduction操作。  

正好Collectors收集器工具类提供了很多静态方法，可以方便的创建收集器实例

List<Employee> emps = Arrays.asList(
    new Employee(101, "Z3", 19, 9999.99),
    new Employee(102, "L4", 20, 7777.77),
    new Employee(103, "W5", 35, 6666.66),
    new Employee(104, "Tom", 44, 1111.11),
    new Employee(105, "Jerry", 60, 4444.44)
);

@Test
public void test02(){
    //放入List
    List<String> list = emps.stream()
        .map(Employee::getName)
        .collect(Collectors.toList()); 
    list.forEach(System.out::println);
    
    //放入Set
    Set<String> set = emps.stream()
        .map(Employee::getName)
        .collect(Collectors.toSet());
    set.forEach(System.out::println);

    //放入LinkedHashSet
    LinkedHashSet<String> linkedHashSet = emps.stream()
        .map(Employee::getName)
        .collect(Collectors.toCollection(LinkedHashSet::new));
    linkedHashSet.forEach(System.out::println);
}

@Test
public void test03(){
    //总数
    Long count = emps.stream()
        .collect(Collectors.counting());
    System.out.println(count);

    //平均值
    //Lambda 参数列表中的第一个参数是方法的调用者，第二个参数是方法的参数时（或不存在时），可以使用 ClassName :: Method
    Double avg = emps.stream()
        .collect(Collectors.averagingDouble(Employee::getSalary));
    System.out.println(avg);

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

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

    //最小值
    Optional<Double> min = emps.stream()
        .map(Employee::getSalary)
        .collect(Collectors.minBy(Double::compare));
    System.out.println(min.get());
}

@Test
public void test04(){
    //分组
    Map<Integer, List<Employee>> map = emps.stream()
        .collect(Collectors.groupingBy(Employee::getId));
    System.out.println(map);

    //多级分组
    Map<Integer, Map<String, List<Employee>>> mapMap = emps.stream()
        .collect(Collectors.groupingBy(Employee::getId, Collectors.groupingBy((e) -> {
            if (e.getAge() > 35) {
                return "开除";
            } else {
                return "继续加班";
            }
        })));
    System.out.println(mapMap);
    
    //分区
    Map<Boolean, List<Employee>> listMap = emps.stream()
        .collect(Collectors.partitioningBy((e) -> e.getSalary() > 4321));
    System.out.println(listMap);
}

@Test
public void test05(){
    //总结
    DoubleSummaryStatistics dss = emps.stream()
        .collect(Collectors.summarizingDouble(Employee::getSalary));
    System.out.println(dss.getMax());
    System.out.println(dss.getMin());
    System.out.println(dss.getSum());
    System.out.println(dss.getCount());
    System.out.println(dss.getAverage());
    
    //连接
    String str = emps.stream()
        .map(Employee::getName)
        .collect(Collectors.joining("-")); //可传入分隔符
    System.out.println(str);
}

其中：Collectors.toCollection(LinkedHashSet::new)

static <T,C extends Collection<T>>
Collector<T,?,C> toCollection(Supplier<C> collectionFactory) 
返回一个 Collector ，按照遇到的顺序将输入元素累加到一个新的 Collection中。  

并行流

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

@Test
public void test03(){
    //串行流(单线程)：切换为并行流 parallel()
    //并行流：切换为串行流 sequential()
    LongStream.rangeClosed(0, 100000000L)
        .parallel() //底层：ForkJoin
        .reduce(0, Long::sum);

}

Fork / Join 框架：

Fork/Join 框架：就是在必要的情况下，将一个大任务，进行拆分（fork） 成若干个小任务（拆到给出的临界值为止），再将一个个的小任务运算的结果 进行join汇总

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

1.采用 “工作窃取” 模式 （work-stealing）

当执行新的任务时它可以将其拆分成 更小的任务执行，并将小任务加到线程队列中，当没有任务执行时，再从一个随机线程的队列中偷一个并把它放在自己的队列中

2.相对于一般的线程池实现 ，fork/join 框架的优势体现在对其中包含的任务的处理方式上，在一般的线程池中，如果一个线程正在执行的任务由于某些原因无法继续运行那么该线程会处于等待状态。

而在fork/join 框架实现中，如果某个子问题由于等待另外一个子问题的完成而无法继续运行。那么处理该子问题的线程会主动寻找其他尚未运行的子问题（窃取过来）来执行，这种方式减少了线程的等待时间，提高了性能

 Fork / Join 实现：

public class ForkJoinCalculate extends RecursiveTask<Long> {

    private static final long serialVersionUID = 1234567890L;

    private long start;
    private long end;

    private static final long THRESHPLD = 10000;

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

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

        if (length <= THRESHPLD) {
            long sum = 0;
            for (long i = start; i <= end; i++) {
                sum += i;
            }
        } else {
            long middle = (start + end) / 2;

            ForkJoinCalculate left = new ForkJoinCalculate(start, end);
            left.fork(); //拆分子任务 压入线程队列

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

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

        return null;
    }
}

public class TestForkJoin {

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

        ForkJoinPool pool = new ForkJoinPool();
        ForkJoinCalculate task = new ForkJoinCalculate(0, 100000000L);

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

        Instant end = Instant.now();
        System.out.println(Duration.between(start, end).getNano());
    }

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

        Long sum = 0L;
        for (long i = 0; i < 100000000L; i++) {
            sum += i;
        }

        Instant end = Instant.now();
        System.out.println(Duration.between(start, end).getNano());
    }
}