浅谈 parallelStream和Stream 源码及其应用场景

上篇讲述了list.forEach()和list.stream().forEach() 异同点
谈到了并行流的概念，本篇则从源码出发，了解一下其原理。

一、流的初始操作流程

jdk8中将Collection中加入了转换流的概念。

default Stream<E> stream() {
        return StreamSupport.stream(spliterator(), false);
    }

default Stream<E> parallelStream() {
        return StreamSupport.stream(spliterator(), true);
    }

在这里插入图片描述
目前看到的两者是一个参数的区别。

//boolean parallel 是否为并行流
public static <T> Stream<T> stream(Spliterator<T> spliterator, boolean parallel) {
		//用于检查传入的spliterator是否为空
        Objects.requireNonNull(spliterator);
        //ReferencePipeline.Head 表示流的开始，根据spliterator以及parallel创建对应的流操作链
        return new ReferencePipeline.Head<>(spliterator,
                                            StreamOpFlag.fromCharacteristics(spliterator),
                                            parallel);
    }

//该构造方法用于初始化Head类的实例
Head(Spliterator<?> source,
             int sourceFlags, boolean parallel) {
            super(source, sourceFlags, parallel);
        }
ReferencePipeline(Spliterator<?> source,
                      int sourceFlags, boolean parallel) {
        //super关键字调用父类的构造方法，完成对父类的初始化工作
        super(source, sourceFlags, parallel);
    }

//按照给定的参数初始化AbstractPipeline类的实例
AbstractPipeline(Spliterator<?> source,
                     int sourceFlags, boolean parallel) {
        this.previousStage = null;
        this.sourceSpliterator = source;  
        this.sourceStage = this;  //当前阶段作为源操作
        this.sourceOrOpFlags = sourceFlags & StreamOpFlag.STREAM_MASK;
        // The following is an optimization of:
        // StreamOpFlag.combineOpFlags(sourceOrOpFlags, StreamOpFlag.INITIAL_OPS_VALUE);
        this.combinedFlags = (~(sourceOrOpFlags << 1)) & StreamOpFlag.INITIAL_OPS_VALUE;  //左移一位取反操作
        this.depth = 0;
        this.parallel = parallel; //当前流水线是否支持并行操作
    }

二、forEach操作

在这里插入图片描述

@Override
        public void forEach(Consumer<? super E_OUT> action) {
            if (!isParallel()) {
            	//串行流执行
                sourceStageSpliterator().forEachRemaining(action);
            }
            else {
            	//并行流执行
                super.forEach(action);
            }
        }

在这里插入图片描述

1）串行流

demo debug add操作，看为何会报错？

public static void main(String[] args) {
        List<String> list= new ArrayList<>();
        list.add("Sunday");
        list.add("Monday");
        list.add("Tuesday");
        list.add("Wednesday");
        list.add("Thursday");
        list.add("Friday");
        list.add("Saturday");
        list.stream().forEach(d->{
            System.out.println("value="+d);
            if (d.equals("Thursday")){
                list.add(d);
            }
        });
    }

//如果此管道截断是源阶段，则获取源阶段拆分器。调用此方法并成功返回后，将消耗管道
final Spliterator<E_OUT> sourceStageSpliterator() {
        if (this != sourceStage)
            throw new IllegalStateException();

        if (linkedOrConsumed)
            throw new IllegalStateException(MSG_STREAM_LINKED);
        linkedOrConsumed = true;

        if (sourceStage.sourceSpliterator != null) {
            @SuppressWarnings("unchecked")
            Spliterator<E_OUT> s = sourceStage.sourceSpliterator;
            sourceStage.sourceSpliterator = null;
            return s;
        }
        else if (sourceStage.sourceSupplier != null) {
            @SuppressWarnings("unchecked")
            Spliterator<E_OUT> s = (Spliterator<E_OUT>) sourceStage.sourceSupplier.get();
            sourceStage.sourceSupplier = null;
            return s;
        }
        else {
            throw new IllegalStateException(MSG_CONSUMED);
        }
    }

forEachRemaining实现方法
在这里插入图片描述

return未走，直接走了异常返回 throw new ConcurrentModificationException();

public void forEachRemaining(Consumer<? super E> action) {
            int i, hi, mc; // hoist accesses and checks from loop
            ArrayList<E> lst; Object[] a;
            if (action == null)
                throw new NullPointerException();
            if ((lst = list) != null && (a = lst.elementData) != null) {
                if ((hi = fence) < 0) {
                    mc = lst.modCount;
                    hi = lst.size;
                }
                else
                    mc = expectedModCount;
                //i表示开始迭代的位置
                //i=index index表示上次迭代的位置，将上次迭代器正在迭代的位置复制给i
                //(i=index)>=0 保证当前迭代的下标大于等于0
                //表示最大迭代到hi,设置最大的hi=a.length
                //(index = hi) <= a.length保证数组不跨界
                if ((i = index) >= 0 && (index = hi) <= a.length) {
                    for (; i < hi; ++i) {
                        @SuppressWarnings("unchecked") E e = (E) a[i];
                        //执行具体的迭代
                        action.accept(e);
                    }
                    if (lst.modCount == mc)
                        return;
                }
            }
            throw new ConcurrentModificationException();
        }

2）并行流

在这里插入图片描述

@Override
    public void forEach(Consumer<? super P_OUT> action) {
        evaluate(ForEachOps.makeRef(action, false));
    }

//构造一个 {@code TerminalOp}，用于对流的每个元素执行操作。
//action，接收流所有元素的 {@code Consumer} 
//ordered,是否请求有序遍历，因为是并行流，所以ordered未false
public static <T> TerminalOp<T, Void> makeRef(Consumer<? super T> action,
                                                  boolean ordered) {
        Objects.requireNonNull(action);
        return new ForEachOp.OfRef<>(action, ordered);
    }

//使用终端操作评估管道以产生结果。
final <R> R evaluate(TerminalOp<E_OUT, R> terminalOp) {
        assert getOutputShape() == terminalOp.inputShape();
        if (linkedOrConsumed)
            throw new IllegalStateException(MSG_STREAM_LINKED);
        linkedOrConsumed = true;

        return isParallel()
               ? terminalOp.evaluateParallel(this, sourceSpliterator(terminalOp.getOpFlags()))
               : terminalOp.evaluateSequential(this, sourceSpliterator(terminalOp.getOpFlags()));
    }

三、核心原理-ForkJoinPool

其核心原理则-ForkJoinPool

1）Diagrams

在这里插入图片描述

2）compute()

ForkJoin进行计算任务时，计算类是要继承ForkJoinTask并且重写compute方法的。
我们看一下ForkJoinTask内部类是如何重写compute()方法的。

//类似于 AbstractTask，但不需要跟踪子任务
public void compute() {
            Spliterator<S> rightSplit = spliterator, leftSplit;
            //先调用当前splititerator 方法的estimateSize 方法，预估这个分片中的数据量
            long sizeEstimate = rightSplit.estimateSize(), sizeThreshold;
            if ((sizeThreshold = targetSize) == 0L)
                targetSize = sizeThreshold = AbstractTask.suggestTargetSize(sizeEstimate);
            boolean isShortCircuit = StreamOpFlag.SHORT_CIRCUIT.isKnown(helper.getStreamAndOpFlags());
            boolean forkRight = false;
            Sink<S> taskSink = sink;
            ForEachTask<S, T> task = this;
            while (!isShortCircuit || !taskSink.cancellationRequested()) {
                if (sizeEstimate <= sizeThreshold ||
                    (leftSplit = rightSplit.trySplit()) == null) {
                    task.helper.copyInto(taskSink, rightSplit);
                    break;
                }
                ForEachTask<S, T> leftTask = new ForEachTask<>(task, leftSplit);
                task.addToPendingCount(1);
                ForEachTask<S, T> taskToFork;
                if (forkRight) {
                    forkRight = false;
                    rightSplit = leftSplit;
                    taskToFork = task;
                    task = leftTask;
                }
                else {
                    forkRight = true;
                    taskToFork = leftTask;
                }
                //根据预估的数据量获取最小处理单元的大小阈值，即当数据量已经小于这个阈值的时候进行计算，否则进行fork.fork方法中调用了ForkJoinPool线程池并行计算
                taskToFork.fork();
                //将任务划分成更小的数据块，进行求解
                sizeEstimate = rightSplit.estimateSize();
            }
            task.spliterator = null;
            task.propagateCompletion();
        }

重写的compute()方法，当进行fork方法时，实际就是调用了ForkJoinPool线程池进行计算了，那么线程池本身是无顺序的，谁先计算完谁展示。

3）ForkJoinPool核心算法

在这里插入图片描述

“工作窃取”（work-stealing）算法

ForkJoinPool的基本原理是基于“工作窃取”（work-stealing）算法。它维护着一个工作队列（WorkQueue）的数组，每个工作队列对应一个工作线程（WorkerThread）。当一个线程需要执行一个任务时，它会将任务添加到自己的工作队列中。当一个线程的工作队列为空时，它会从其他线程的工作队列中“窃取”一个任务来执行。这个“窃取”操作可以在不同的线程间实现任务的负载均衡。
工作窃取算法的优点是充分利用线程进行并行计算，并减少了线程间的竞争，其缺点是在某些情况下还是存在竞争，比如双端队列里只有一个任务时。并且消耗了更多的系统资源，比如创建多个线程和多个双端队列。

“分治法”(Divide-and-Conquer Algorithm)

分治法-典型的应用比如快速排序算法。使用分治法来实现任务的并行执行。分治法是一种将大问题划分成小问题，并通过递归地解决小问题来解决大问题的方法。

四、结果无顺序

在这里插入图片描述

1）若想并行且有顺序，用.forEachOrdered替代

在这里插入图片描述

2）.forEachOrdered是如何保证有序的？

private static <S, T> void doCompute(ForEachOrderedTask<S, T> task) {
            Spliterator<S> rightSplit = task.spliterator, leftSplit;
            long sizeThreshold = task.targetSize;
            boolean forkRight = false;
            while (rightSplit.estimateSize() > sizeThreshold &&
                   (leftSplit = rightSplit.trySplit()) != null) {
                ForEachOrderedTask<S, T> leftChild =
                    new ForEachOrderedTask<>(task, leftSplit, task.leftPredecessor);
                ForEachOrderedTask<S, T> rightChild =
                    new ForEachOrderedTask<>(task, rightSplit, leftChild);

                // 分叉父任务 完成左右子项 “发生在”父项完成之前
                task.addToPendingCount(1);
                // 完成左边的孩子“发生在”完成右边的孩子之前
                rightChild.addToPendingCount(1);
                task.completionMap.put(leftChild, rightChild);

                // If task is not on the left spine
                if (task.leftPredecessor != null) {
                    /*
                     * Completion of left-predecessor, or left subtree,
                     * "happens-before" completion of left-most leaf node of
                     * right subtree.
                     * The left child's pending count needs to be updated before
                     * it is associated in the completion map, otherwise the
                     * left child can complete prematurely and violate the
                     * "happens-before" constraint.
                     */
                    leftChild.addToPendingCount(1);
                    // Update association of left-predecessor to left-most
                    // leaf node of right subtree
                    if (task.completionMap.replace(task.leftPredecessor, task, leftChild)) {
                        // If replaced, adjust the pending count of the parent
                        // to complete when its children complete
                        task.addToPendingCount(-1);
                    } else {
                        // Left-predecessor has already completed, parent's
                        // pending count is adjusted by left-predecessor;
                        // left child is ready to complete
                        leftChild.addToPendingCount(-1);
                    }
                }

                ForEachOrderedTask<S, T> taskToFork;
                if (forkRight) {
                    forkRight = false;
                    rightSplit = leftSplit;
                    task = leftChild;
                    taskToFork = rightChild;
                }
                else {
                    forkRight = true;
                    task = rightChild;
                    taskToFork = leftChild;
                }
                taskToFork.fork();
            }

            /*
             * Task's pending count is either 0 or 1.  If 1 then the completion
             * map will contain a value that is task, and two calls to
             * tryComplete are required for completion, one below and one
             * triggered by the completion of task's left-predecessor in
             * onCompletion.  Therefore there is no data race within the if
             * block.
             */
            if (task.getPendingCount() > 0) {
                // Cannot complete just yet so buffer elements into a Node
                // for use when completion occurs
                @SuppressWarnings("unchecked")
                IntFunction<T[]> generator = size -> (T[]) new Object[size];
                Node.Builder<T> nb = task.helper.makeNodeBuilder(
                        task.helper.exactOutputSizeIfKnown(rightSplit),
                        generator);
                task.node = task.helper.wrapAndCopyInto(nb, rightSplit).build();
                task.spliterator = null;
            }
            task.tryComplete();
        }

看代码会大概得理解，此采用了“happens-before”原则，左子树需右子树之前完成，通过计数策略保证前后顺序的完成，继而保证了其有序性。最终执行依旧是ForkJoinPool线程池执行。
在这里插入图片描述

五、应用场景

1）并行的前提是需要硬件支持

前提是硬件支持, 如果单核 CPU, 只会存在并发处理, 而不会并行

2）demo 测试性能（本机测试）

测试配置：16 GB 2667 MHz DDR4

@Test
    public void dateTest() {
        System.out.println("数据汇总开始");
        Long startTime = System.currentTimeMillis();
        int count1 = adminTaskReceiveService.receiveCount();
        int count2 = adminTaskReceiveService.inspectCount();
        int count3=adminTaskReceiveService.constructCount();
        int count4=adminTaskReceiveService.appointmentCount();
        TestResult testResult = new TestResult();
        testResult.setReceiveCount(count1);
        testResult.setInspectCount(count2);
        testResult.setConstructCount(count3);
        testResult.setAppointmentCount(count4);

        int count11 = adminTaskReceiveService.receiveCount1();
        int count22 = adminTaskReceiveService.inspectCount1();
        int count33=adminTaskReceiveService.constructCount1();
        int count44=adminTaskReceiveService.appointmentCount1();
        testResult.setReceiveCount1(count11);
        testResult.setInspectCount1(count22);
        testResult.setConstructCount1(count33);
        testResult.setAppointmentCount1(count44);

        System.out.println("数据汇总结束，result=" + testResult);
        Long endTime = System.currentTimeMillis();
        System.out.println("time=" + (endTime - startTime) + "毫秒");
    }

    @Test
    public void dateTest1() {
        System.out.println("数据汇总开始");
        Long startTime = System.currentTimeMillis();
        TestResult testResult = new TestResult();
        List<Runnable> taskList = new ArrayList<Runnable>() {
            {
                add(() -> testResult.setReceiveCount(adminTaskReceiveService.receiveCount()));
                add(() -> testResult.setInspectCount(adminTaskReceiveService.inspectCount()));
                add(() -> testResult.setConstructCount(adminTaskReceiveService.constructCount()));
                add(() -> testResult.setAppointmentCount(adminTaskReceiveService.appointmentCount()));

                add(() -> testResult.setReceiveCount1(adminTaskReceiveService.receiveCount1()));
                add(() -> testResult.setInspectCount1(adminTaskReceiveService.inspectCount1()));
                add(() -> testResult.setConstructCount1(adminTaskReceiveService.constructCount1()));
                add(() -> testResult.setAppointmentCount1(adminTaskReceiveService.appointmentCount1()));
            }
        };
        taskList.parallelStream().forEach(Runnable::run);
        System.out.println("数据汇总结束，result=" + testResult);
        Long endTime = System.currentTimeMillis();
        System.out.println("time=" + (endTime - startTime) + "毫秒");
    }

一个单线程，一个并行，看结果：测试demo我是需要统计8个数量，由结果可见性能并没什么大区别。
由结果可知：并行处理并不总是能提高性能，特别是当任务规模较小或者任务之间依赖性较强时。此外，在使用并行流时，应该避免使用会修改原始集合的操作，因为这些操作可能会导致不可预测的结果。由于’foreach`操作是终端操作，它会阻塞主线程直到所有元素都被处理完毕，因此即使操作是并行的，它们仍然是按照顺序完成的。
在这里插入图片描述