java8流式编程--Java8 Stream＜T＞ 接口

Java8的Stream对一个元素序列提供顺序和并行运算。最重要的4个能力是：

1. 支持函数式编程
2. 提供管道运算能力
3. 提供并发计算能力
4. 提供大量的操作

关联面试题：

1. 什么是流？
2. ::运算符的作用？
3. Java 8的Stream价值是什么？
4. 创建Stream有几种方法？
5. coding:利用parallel并发执行任务？

流是随着时间产生的数据序列。 我们写程序的时候还有很多种流。比如文件流、字符流、网络流、信息流、订阅流……
Stream也是流，它为序列提供顺序、并发运算。

管道是组织流计算的一种手段，Linux的管道连接一个进程的输出到另一个进程的输入。Stream底层也是管道在支持。每一步操作，都是一个管道。这个管道。连接上游的输出，和下游的输入。Stream的管道设计。是从数据源开始连接上很多中间操作(intermediate operation)，对数据进行变换。最后接上一个终止操作（terminal operation)对结果进行运算。和转换。

@Test
public void test_mapfilter() {
    Stream.of(1,2,3,4,5,6)
        .map(x -> x.toString())
        .map(x -> x + x)
        .map(x -> x + x + x)
        //                .map(x -> Integer.parseInt(x))
        .map(Integer::parseInt)
        .forEach(x -> {
            System.out.println(x);
        });
}

Stream.of是一个数据源的工厂，用来构造流。除此之外容器(collection)类型都可以用stream() 方法直接构造流。

函数式编程最关心的就是类型。map将一个类型映射成另一个类型。forEach是终止操作。

状态是一个很宽泛的概念。比如内存中的值、寄存器中的值发生变化我们就称之为状态的变化。比如说变量的数据、类型的成员都可以称之为状态。

下面的程序中，sorted是一个中间操作。他对原始数据进行排序，内部的实现中sorted 会改变原始数据的顺序。这种我们称之为stateful ，也就是有状态操作。

@Test
public void test_mutation() {
    var stream = Stream.of(1,3,5,2,3,4,5,6).sorted();
    stream.forEach(System.out::println);
}

相比之下，map操作，不改变任何状态。只是将一个类型映射成另一个类型，我们称之为无状态操（stateless）

@Test
public void test_flatMap(){
    // String -> Stream
    var set = Stream.of("My", "Mine")
        .flatMap(str -> str.chars().mapToObj(i -> (char)i))
        .collect(Collectors.toSet());
    System.out.println(set.stream().collect(Collectors.toList()));
}

并行、并发等概念会在并发编程部分详细介绍。

流经过parallel或者parallelStream运算之后，内部会创建ForkJoinPool组织并行计算。通常是以CPU核数-1数量创建线程。也可以自己提供ForkJoinPool。

下面程序中我们自己提供了ForkJoinPool，约定线程数为2

@Test
public void test_parallel() throws ExecutionException, InterruptedException {

    var r = new Random();
    var list = IntStream.range(0, 1_000_000)
        .map(x -> r.nextInt(10_000_000))
        .boxed()
        .collect(Collectors.toList());

    var t0 = System.currentTimeMillis();
    System.out.println(list.stream().max((a, b) -> a - b));
    System.out.println("time:" + (System.currentTimeMillis() - t0));

    // 1000
    var pool = new ForkJoinPool(2);
    var t1 = System.currentTimeMillis();
    var max = pool.submit(() -> list.parallelStream().max((a , b) -> a -b)).get();

    //        list.stream().parallel().max((a , b) -> a -b);

    // 15
    // Spliter 1024 -> Thread0 1024 -> Thread1

    System.out.println("time:" + (System.currentTimeMillis() - t1) + ",max:" + max);
}