J.U.C Review - Stream并行计算原理源码分析

2024-09-07 06:52

本文主要是介绍J.U.C Review - Stream并行计算原理源码分析,希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!

文章目录

  • Java 8 Stream简介
  • Stream单线程串行计算
  • Stream多线程并行计算
  • 源码分析Stream并行计算原理
  • Stream并行计算的性能提升

在这里插入图片描述

Java 8 Stream简介

自Java 8推出以来,开发者可以使用Stream接口和lambda表达式实现流式计算。这种编程风格不仅简化了对集合操作的代码,还提高了代码的可读性和性能。

Stream接口提供了多种集合操作方法,包括empty(判空)、filter(过滤)、max(求最大值)、findFirstfindAny(查找操作)等,使得对集合的操作更加灵活和直观。


Stream单线程串行计算

在默认情况下,Stream接口是以串行的方式运行的,这意味着所有的操作都在一个线程内执行。我们可以通过以下示例代码展示这一点:

public class StreamDemo {public static void main(String[] args) {Stream.of(1, 2, 3, 4, 5, 6, 7, 8, 9).reduce((a, b) -> {System.out.println(String.format("%s: %d + %d = %d",Thread.currentThread().getName(), a, b, a + b));return a + b;}).ifPresent(System.out::println);}
}

在这个例子中,我们通过Stream.of()方法创建了一个包含数字1到9的流。随后,调用reduce方法对这些数字进行累加操作。reduce方法的作用是从前两个元素开始,执行指定操作(在此示例中为加法),然后将结果与下一个元素进行相同的操作,直到处理完所有元素。

程序的输出如下:

main: 1 + 2 = 3  
main: 3 + 3 = 6  
main: 6 + 4 = 10  
main: 10 + 5 = 15  
main: 15 + 6 = 21  
main: 21 + 7 = 28  
main: 28 + 8 = 36  
main: 36 + 9 = 45  
45

从输出可以看出,所有计算均由main线程执行,并且操作是严格按照元素顺序串行完成的。


Stream多线程并行计算

然而,单线程串行执行并不是唯一的选择。在现代多核处理器的时代,我们可以通过并行计算来更高效地利用计算资源。例如,当计算1+2=3的同时,我们可以在另一个线程中计算3+4=7,最后将这些部分结果进行合并。这种思想与Fork/Join框架的设计理念非常类似。

通过以下代码,我们可以让Stream在多线程中并行执行:

public class StreamParallelDemo {public static void main(String[] args) {Stream.of(1, 2, 3, 4, 5, 6, 7, 8, 9).parallel().reduce((a, b) -> {System.out.println(String.format("%s: %d + %d = %d",Thread.currentThread().getName(), a, b, a + b));return a + b;}).ifPresent(System.out::println);}
}

运行这段代码,输出如下:

ForkJoinPool.commonPool-worker-1: 3 + 4 = 7  
ForkJoinPool.commonPool-worker-4: 8 + 9 = 17  
ForkJoinPool.commonPool-worker-2: 5 + 6 = 11  
ForkJoinPool.commonPool-worker-3: 1 + 2 = 3  
ForkJoinPool.commonPool-worker-4: 7 + 17 = 24  
ForkJoinPool.commonPool-worker-4: 11 + 24 = 35  
ForkJoinPool.commonPool-worker-3: 3 + 7 = 10  
ForkJoinPool.commonPool-worker-3: 10 + 35 = 45  
45

从输出结果可以看出,这些计算是并行完成的,使用了ForkJoinPool中的commonPool线程池。尽管各个部分的计算是并行执行的,最终的结果仍然是正确的,因为Fork/Join框架负责协调这些并行任务。


源码分析Stream并行计算原理

通过以上的实践,我们知道Stream的并行计算底层是基于Fork/Join框架的。但具体是如何实现的?我们可以通过源码分析来探究。

首先,Stream.of()方法只是生成一个简单的流。接下来,我们查看parallel()方法的实现。由于这里的数据类型是int,因此调用的是BaseStream接口的parallel()方法。BaseStream接口的唯一实现类是AbstractPipeline类。以下是AbstractPipeline类的parallel()方法:

public final S parallel() {sourceStage.parallel = true;return (S) this;
}

这个方法的作用非常简单,仅仅是将sourceStage.parallel标志位设置为true,表示该流将以并行方式执行。

接下来,查看reduce方法的实现。Stream.reduce()方法的具体实现是通过ReferencePipeline这个抽象类,该类继承了AbstractPipeline类:

@Override
public final Optional<P_OUT> reduce(BinaryOperator<P_OUT> accumulator) {return evaluate(ReduceOps.makeRef(accumulator));
}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()));
}@Override
public final boolean isParallel() {return sourceStage.parallel;
}

从源码可以看出,reduce方法调用了evaluate方法,而evaluate方法根据parallel标志位来决定是并行执行还是串行执行。如果paralleltrue,则调用evaluateParallel方法,否则调用evaluateSequential方法。

我们再来看evaluateParallel方法在ReduceOps.ReduceOp类中的具体实现:

@Override
public <P_IN> R evaluateParallel(PipelineHelper<T> helper,Spliterator<P_IN> spliterator) {return new ReduceTask<>(this, helper, spliterator).invoke().get();
}

evaluateParallel方法创建了一个ReduceTask实例,并调用其invoke()方法来执行计算。ReduceTask类继承自AbstractTaskAbstractTask又继承自CountedCompleter,最终继承自ForkJoinTask。这就解释了为什么Stream的并行计算底层使用了Fork/Join框架。


Stream并行计算的性能提升

最后,我们通过一个简单的性能测试来验证Stream并行计算的优势。下面的代码演示了如何计算一千万个随机数的和,并比较串行计算和并行计算的时间开销:

public class StreamParallelDemo {public static void main(String[] args) {System.out.println(String.format("本计算机的核数:%d", Runtime.getRuntime().availableProcessors()));Random random = new Random();List<Integer> list = new ArrayList<>(1000_0000);for (int i = 0; i < 1000_0000; i++) {list.add(random.nextInt(100));}long prevTime = getCurrentTime();list.stream().reduce((a, b) -> a + b).ifPresent(System.out::println);System.out.println(String.format("单线程计算耗时:%d", getCurrentTime() - prevTime));prevTime = getCurrentTime();list.stream().parallel().reduce((a, b) -> a + b).ifPresent(System.out::println);System.out.println(String.format("多线程计算耗时:%d", getCurrentTime() - prevTime));}private static long getCurrentTime() {return System.currentTimeMillis();}
}

在一台8核计算机上的输出结果如下:

本计算机的核数:8  
495156156  
单线程计算耗时:223  
495156156  
多线程计算耗时:95  

结果表明,在多核环境下,Stream的并行计算相比串行计算确实能够显著提升性能。然而,性能提升的幅度并非线性增长,因为线程管理和上下文切换本身也会带来一定的开销。如果在单核环境中,串行计算反而可能会比并行计算更快。

总结而言,Java 8的Stream并行计算通过简化代码的方式,利用了底层的多核资源,大幅提升了复杂集合操作的性能。然而在实际应用中,开发者需要根据具体的硬件环境和任务特性来决定是否使用并行计算。

在这里插入图片描述

这篇关于J.U.C Review - Stream并行计算原理源码分析的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!



http://www.chinasem.cn/article/1144375

相关文章

Java程序进程起来了但是不打印日志的原因分析

《Java程序进程起来了但是不打印日志的原因分析》:本文主要介绍Java程序进程起来了但是不打印日志的原因分析,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教... 目录Java程序进程起来了但是不打印日志的原因1、日志配置问题2、日志文件权限问题3、日志文件路径问题4、程序

Java 正则表达式URL 匹配与源码全解析

《Java正则表达式URL匹配与源码全解析》在Web应用开发中,我们经常需要对URL进行格式验证,今天我们结合Java的Pattern和Matcher类,深入理解正则表达式在实际应用中... 目录1.正则表达式分解:2. 添加域名匹配 (2)3. 添加路径和查询参数匹配 (3) 4. 最终优化版本5.设计思

Java字符串操作技巧之语法、示例与应用场景分析

《Java字符串操作技巧之语法、示例与应用场景分析》在Java算法题和日常开发中,字符串处理是必备的核心技能,本文全面梳理Java中字符串的常用操作语法,结合代码示例、应用场景和避坑指南,可快速掌握字... 目录引言1. 基础操作1.1 创建字符串1.2 获取长度1.3 访问字符2. 字符串处理2.1 子字

Spring Boot循环依赖原理、解决方案与最佳实践(全解析)

《SpringBoot循环依赖原理、解决方案与最佳实践(全解析)》循环依赖指两个或多个Bean相互直接或间接引用,形成闭环依赖关系,:本文主要介绍SpringBoot循环依赖原理、解决方案与最... 目录一、循环依赖的本质与危害1.1 什么是循环依赖?1.2 核心危害二、Spring的三级缓存机制2.1 三

C#中async await异步关键字用法和异步的底层原理全解析

《C#中asyncawait异步关键字用法和异步的底层原理全解析》:本文主要介绍C#中asyncawait异步关键字用法和异步的底层原理全解析,本文给大家介绍的非常详细,对大家的学习或工作具有一... 目录C#异步编程一、异步编程基础二、异步方法的工作原理三、代码示例四、编译后的底层实现五、总结C#异步编程

Python 迭代器和生成器概念及场景分析

《Python迭代器和生成器概念及场景分析》yield是Python中实现惰性计算和协程的核心工具,结合send()、throw()、close()等方法,能够构建高效、灵活的数据流和控制流模型,这... 目录迭代器的介绍自定义迭代器省略的迭代器生产器的介绍yield的普通用法yield的高级用法yidle

Go 语言中的select语句详解及工作原理

《Go语言中的select语句详解及工作原理》在Go语言中,select语句是用于处理多个通道(channel)操作的一种控制结构,它类似于switch语句,本文给大家介绍Go语言中的select语... 目录Go 语言中的 select 是做什么的基本功能语法工作原理示例示例 1:监听多个通道示例 2:带

鸿蒙中@State的原理使用详解(HarmonyOS 5)

《鸿蒙中@State的原理使用详解(HarmonyOS5)》@State是HarmonyOSArkTS框架中用于管理组件状态的核心装饰器,其核心作用是实现数据驱动UI的响应式编程模式,本文给大家介绍... 目录一、@State在鸿蒙中是做什么的?二、@Spythontate的基本原理1. 依赖关系的收集2.

C++ Sort函数使用场景分析

《C++Sort函数使用场景分析》sort函数是algorithm库下的一个函数,sort函数是不稳定的,即大小相同的元素在排序后相对顺序可能发生改变,如果某些场景需要保持相同元素间的相对顺序,可使... 目录C++ Sort函数详解一、sort函数调用的两种方式二、sort函数使用场景三、sort函数排序

Java调用C++动态库超详细步骤讲解(附源码)

《Java调用C++动态库超详细步骤讲解(附源码)》C语言因其高效和接近硬件的特性,时常会被用在性能要求较高或者需要直接操作硬件的场合,:本文主要介绍Java调用C++动态库的相关资料,文中通过代... 目录一、直接调用C++库第一步:动态库生成(vs2017+qt5.12.10)第二步:Java调用C++