浅谈 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操作

在这里插入图片描述

@Overridepublic 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 loopArrayList<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;}elsemc = 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）并行流

在这里插入图片描述

@Overridepublic 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 spineif (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 subtreeif (task.completionMap.replace(task.leftPredecessor, task, leftChild)) {// If replaced, adjust the pending count of the parent// to complete when its children completetask.addToPendingCount(-1);} else {// Left-predecessor has already completed, parent's// pending count is adjusted by left-predecessor;// left child is ready to completeleftChild.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

@Testpublic 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) + "毫秒");}@Testpublic 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`操作是终端操作，它会阻塞主线程直到所有元素都被处理完毕，因此即使操作是并行的，它们仍然是按照顺序完成的。
在这里插入图片描述