Java8 中的 Stream 那么强大，那你知道它的原理是什么吗？

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

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

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

本文会对Stream的实现原理进行剖析。

Stream的组成与特点

• 元素是特定类型的对象，形成一个队列。 Java中的Stream并_不会_向集合那样存储和管理元素，而是按需计算

• 数据源流的来源可以是集合Collection、数组Array、I/O channel， 产生器generator 等

• 聚合操作类似SQL语句一样的操作， 比如filter, map, reduce, find, match, sorted等

• Pipelining: 中间操作都会返回流对象本身。这样多个操作可以串联成一个管道， 如同流式风格（fluent style）。这样做可以对操作进行优化， 比如延迟执行(laziness evaluation)和短路( short-circuiting)

• 内部迭代：以前对集合遍历都是通过Iterator或者For-Each的方式, 显式的在集合外部进行迭代， 这叫做外部迭代。 Stream提供了内部迭代的方式， 通过访问者模式 (Visitor)实现。

1.0-1.4 中的 java.lang.Thread

5.0 中的 java.util.concurrent

6.0 中的 Phasers 等

7.0 中的 Fork/Join 框架

8.0 中的 Lambda

List numbers = Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9);
numbers.parallelStream()
       .forEach(out::println); 

List numbers = Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9);
numbers.parallelStream()
       .forEachOrdered(out::println);  

BaseStream接口

public interface BaseStream<T, S extends BaseStream<T, S>>
        extends AutoCloseable {
    Iterator iterator();

    Spliterator spliterator();

    boolean isParallel();

    S sequential();

    S parallel();

    S unordered();

    S onClose(Runnable closeHandler);

    void close();
}

S extends BaseStream

如果是并行一定涉及到对当前流的拆分，即将一个流拆分成多个子流，子流肯定和父流的类型是一致的。子流可以继续拆分子流，一直拆分下去…

Stream接口

public interface Stream extends BaseStream> 

Stream filter(Predicatesuper T> predicate);
 Stream map(Functionsuper T, ? extends R> mapper);
 Stream flatMap(Functionsuper T, ? extends Stream> mapper);
Stream sorted();
Stream peek(Consumersuper T> action);
Stream limit(long maxSize);
Stream skip(long n);
...

关闭流操作

/** * Returns an equivalent stream with an additional close handler. Close * handlers are run when the {@link #close()} method * is called on the stream, and are executed in the order they were * added. All close handlers are run, even if earlier close handlers throw * exceptions. If any close handler throws an exception, the first * exception thrown will be relayed to the caller of {@code close()}, with * any remaining exceptions added to that exception as suppressed exceptions * (unless one of the remaining exceptions is the same exception as the * first exception, since an exception cannot suppress itself.) May * return itself. * * 
This is an intermediate * operation. * * @param closeHandler A task to execute when the stream is closed * @return a stream with a handler that is run if the stream is closed */
S onClose(Runnable closeHandler);

• onClose() 方法会返回流对象本身，也就是说可以对改对象进行多次调用

• 如果调用了多个onClose() 方法，它会按照调用的顺序触发，但是如果某个方法有异常则只会向上抛出第一个异常

• 前一个 onClose() 方法抛出了异常不会影响后续 onClose() 方法的使用

• 如果多个 onClose() 方法都抛出异常，只展示第一个异常的堆栈，而其他异常会被压缩，只展示部分信息

并行流和串行流

Returns an equivalent stream that is parallel. May return

itself, either because the stream was already parallel, or because

the underlying stream state was modified to be parallel.

stream.parallel()
   .filter(...)
   .sequential()
   .map(...)
   .parallel()
   .sum();

ParallelStream背后的男人：ForkJoinPool

问题的关键在于，对于一个任务而言，只有当它所有的子任务完成之后，它才能够被执行，想象一下归并排序的过程。

1. 每个工作线程都有自己的工作队列WorkQueue；
2. 这是一个双端队列dequeue，它是线程私有的；
3. ForkJoinTask中fork的子任务，将放入运行该任务的工作线程的队头，工作线程将以LIFO的顺序来处理工作队列中的任务，即堆栈的方式；
4. 为了最大化地利用CPU，空闲的线程将从其它线程的队列中「窃取」任务来执行
5. 但是是从工作队列的尾部窃取任务，以减少和队列所属线程之间的竞争；
6. 双端队列的操作：push()/pop()仅在其所有者工作线程中调用，poll()是由其它线程窃取任务时调用的；
7. 当只剩下最后一个任务时，还是会存在竞争，是通过CAS来实现的；

用ForkJoinPool的眼光来看ParallelStream

List userInfoList =
        DaoContainers.getUserInfoDAO().queryAllByList(new UserInfoModel());
userInfoList.parallelStream().forEach(RedisUserApi::setUserIdUserInfo);

public static String query(String question) {
  List engines = new ArrayList();
  engines.add("http://www.google.com/?q=");
  engines.add("http://duckduckgo.com/?q=");
  engines.add("http://www.bing.com/search?q=");
   
  // get element as soon as it is available
  Optional result = engines.stream().parallel().map((base) - {
    String url = base + question;
    // open connection and fetch the result
    return WS.url(url).get();
  }).findAny();
  return result.get();
}

• 这个并行流计算操作将由主线程和JVM默认的ForkJoinPool.commonPool()来共同执行。

• map中是一个阻塞方法，需要通过访问HTTP接口并得到它的response，所以任何一个worker线程在执行到这里的时候都会阻塞并等待结果。

• 所以当此时再其他地方通过并行流方式调用计算方法的时候，将会受到此处阻塞等待的方法的影响。

• 目前的ForkJoinPool的实现并未考虑补偿等待那些阻塞在等待新生成的线程的工作worker线程，所以最终ForkJoinPool.commonPool()中的线程将备用光并且阻塞等待。

正如我们上面那个列子的情况分析得知，lambda的执行并不是瞬间完成的,所有使用parallel streams的程序都有可能成为阻塞程序的源头， 并且在执行过程中程序中的其他部分将无法访问这些workers，这意味着任何依赖parallel streams的程序在什么别的东西占用着common ForkJoinPool时将会变得不可预知并且暗藏危机。

1. 当需要处理递归分治算法时，考虑使用ForkJoinPool。

2. 仔细设置不再进行任务划分的阈值，这个阈值对性能有影响。

3. Java 8中的一些特性会使用到ForkJoinPool中的通用线程池。在某些场合下，需要调整该线程池的默认的线程数量

4. lambda应该尽量避免副作用，也就是说，避免突变基于堆的状态以及任何IO

5. lambda应该互不干扰，也就是说避免修改数据源（因为这可能带来线程安全的问题）

6. 避免访问在流操作生命周期内可能会改变的状态

并行流的性能

• 数据大小：数据够大，每个管道处理时间够长，并行才有意义；

• 源数据结构：每个管道操作都是基于初始数据源，通常是集合，将不同的集合数据源分割会有一定消耗；

• 装箱：处理基本类型比装箱类型要快；

• 核的数量：默认情况下，核数量越多，底层fork/join线程池启动线程就越多；

• 单元处理开销：花在流中每个元素身上的时间越长，并行操作带来的性能提升越明显；

• 性能好：ArrayList、数组或IntStream.range(数据支持随机读取，能轻易地被任意分割)

• 性能一般：HashSet、TreeSet(数据不易公平地分解，大部分也是可以的)

• 性能差：LinkedList(需要遍历链表，难以对半分解)、Stream.iterate和BufferedReader.lines(长度未知，难以分解)

NQ模型

遇到顺序

什么时候该使用并行流

References

• http://movingon.cn/2017/05/02/jdk8-Stream-BaseStream-%E6%BA%90%E7%A0%81%E9%9A%BE%E7%82%B9%E6%B5%85%E6%9E%901/
• https://www.jianshu.com/p/bd825cb89e00
• https://jrebel.com/rebellabs/java-parallel-streams-are-bad-for-your-health/
• https://blog.csdn.net/weixx3/article/details/81266552
• https://www.ibm.com/developerworks/cn/java/j-java-streams-5-brian-goetz/index.html
• https://www.ibm.com/developerworks/cn/java/j-java-streams-3-brian-goetz/index.html
• https://juejin.im/post/5dc5a148f265da4d4f65c191
• https://stackoverrun.com/cn/q/10341100

面试官：Java 反射是什么？
Java 17：和遗留 25 年的漏洞 Say Goodbye
Java 框架 Mybatis 插件开发指南，超详细！
18 个 Java8 处理日期的新花样，肯定没用过！
你还在 new 对象吗？Java8 通用 Builder 了解一下