jdk8 stream与parallelStream实现源码阅读

大家好，欢迎来到IT知识分享网。

从java 8以后，jdk底层新增了Stream API，使我们写出更简洁、干净的代码来对集合、数组等进行操作，先简单介绍下流的几个概念：

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

• 元素是特定类型的对象，形成一个队列。 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)实现。

下面首先看一下jdk源码中如何实现返回流对象（Stream），打开Collection接口源码，可以Collection增加了以下两个方法：

其中可以看到，多出了我们经常使用的两个方法，stream和parallelStream方法，都返回了该集合的Stream对象（该接口中包含了许多操作流的方法，这里不一一列举，详细用法看API，在这里主要阅读源码理解其实现），一看名称，就能大概猜出来其中一个是串行流，另一个是并行流。并且接口中给了这两个方法的默认实现。其中采用了StreamSupport的stream方法，这个类也是java 8新增的一个底层创建和操作Stream的工具类。

那这个StreamSupport中又是做了些啥呢？继续跟代码下去就会发现，它根据提供的元素分割器（Spliterator）和是否并行创建了一个Stream的实现类，ReferencePipeline的静态内部实现类Head类，这个类重写了ReferencePipeline类的几个方法，其中我们主要关注下图两个方法：

forEach和forEachOrderd方法，可以看出其实现大概相同，都是先判断了该Stream是否是并行流，不是的话直接执行原分割迭代器的forEachRemaining方法，对该流的元素进行一一消费处理，是的话则调用父类（ReferencePipeline）的对应处理方法，以其中forEach为例，ReferencePipeline类中的forEach实现如下：

传入了TerminalOp的实现ForEachOp，其中TerminalOp有多种实现，其中包含了下图实现类：

forEach实现类为ForEachOp的OfRef类，其中定义了对单个元素的Consumer,其中两者最大的不同为，并行流调用了evaluateParallel方法，而串行流调用了evaluateSequential方法,两个方法的实现如下：

其中串行流的forEach方式源码实现如下：

可以看到关键方法还是在本线程中调用wrappedSink（为Consumer的实现类）的forEachRemaining或forEachWithCancel方法依次消费流中的元素。

并行流中相对复杂，分为有序和无序分别实现了两个基于CountedCompleter的实现类ForEachOrderedTask和ForEachTask，并调用了其共同父类ForkJoinTask的invoke方法，下图来看看这两个invoke方法中都做了些什么：

可以看到invoke方法中调用了执行方法doInvoke，doInvoke方法中又调用了doExec方法同步执行CountedCompleter的exec方法，其中执行了一个compute抽象方法并返回了false，我们可以在CountedCompleter其中一个实现中，找到compute方法的实现，下面以ForEachTask为例继续阅读：

可以看到，其中调用了ForkJoinTask的fork方法，这就应该很熟悉了，这是ForkJoinPool线程池API推荐的用法，用来并行计算的线程池：

该方法中，首先判断当前线程是不是ForkJoinWorkerThread，如果是，直接将该任务加入到本线程的线程的ForkJoinPool.WorkQueue中，交给本线程引用ForkJoinPool线程池执行。否则交给ForkJoinPool自带的common公用线程池执行。

其中common的最小划分为1（等同于核心线程数），最大线程数为当前系统可用核心数-1个（–除非超过32767，应该很少见吧），读到这里可以看出，系统中所有parallelStream流执行方法底层都是通过ForkJoinPool并行计算来提高效率的，我们也可以通过自己手动创建ForkJoinTask的任务来提交到自定义ForkJoinPool线程池，合理分配线程达到提高单个集合计算的效率。

读完了stream和parallelStream的实现区别，再回过头来，看StreamSupport中的第一个传参Spliterator，数据源分割迭代处理器器，这是java 8为了支持源数据的流操作抽象出来的一个接口，其中主要包含了以下几个方法：

tryAdvance方法取源数据中剩下的元素，存在交给consumer消费；forEachRemaining顺序迭代剩下元素并交给consumer消费；trySplit方法将集合划分方法；estimateSize为返回大概的剩余元素个数；getExactSizeIfKnown返回确切的元素个数，当Spliterator是SIZED的时候；characteristics返回该characteristics具有的特征，比如SIZED，ORDERD等；hasCharacteristics是否包含某个特征；getComparator返回比较器，当Spliterator是ORDERD的时候。了解了这个接口的方法的基本作用，就可以灵活的实现自定义数据源生成器的分割迭代器了。

下面我们可以看下ArrayList实现分割迭代器的类ArrayListSpliterator：

可以看出其中trySplit实现了二分法；tryAdvance直接调用了当前index的元素Consumer的accept方法，并将index+1；forEachRemaining直接使用了数组迭代的随机访问方法。

阅读后小结：

     1、自定义stream实现的关键在于实现自己的Spliterator，然后调用StreamSupport的stream方法即可实现自定义的stream运算。

     2、串行stream目前从ArrayList本身看来并不会带来明显性能提升，比起外部自随机访问法迭代需要复制源数据元素，创建stream对象会带来一定的性能消耗（不排除jdk中其他集合有更优的实现或jdk后期优化）；在数据量相对较多时，可以采用并行stream，多核机器下多线程处理数据可以提高处理效率，但需要注意如果不自定义ForkJoinPool线程池，系统所有并行stream处理都用同一个common线程池，可能会降低某些集合处理速度，增大响应时间。

    3、stream操作提供了常用集合操作，可以极大的精简代码量。

    4、并行流的实现原理为ForkJoinPool，需要保证每个stream是无状态，数据源中各个元素相互无依赖的，返回join结果时为多线程操作，则调用返回的集合也必须是线程安全的。

    5、看上图ArrayList的forEachRemaining实现，stream可以修改数据源。