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编译生成多个.class文件的原理和作用

《Java编译生成多个.class文件的原理和作用》作为一名经验丰富的开发者,在Java项目中执行编译后,可能会发现一个.java源文件有时会产生多个.class文件,从技术实现层面详细剖析这一现象... 目录一、内部类机制与.class文件生成成员内部类(常规内部类)局部内部类(方法内部类)匿名内部类二、

Go标准库常见错误分析和解决办法

《Go标准库常见错误分析和解决办法》Go语言的标准库为开发者提供了丰富且高效的工具,涵盖了从网络编程到文件操作等各个方面,然而,标准库虽好,使用不当却可能适得其反,正所谓工欲善其事,必先利其器,本文将... 目录1. 使用了错误的time.Duration2. time.After导致的内存泄漏3. jsO

Python中随机休眠技术原理与应用详解

《Python中随机休眠技术原理与应用详解》在编程中,让程序暂停执行特定时间是常见需求,当需要引入不确定性时,随机休眠就成为关键技巧,下面我们就来看看Python中随机休眠技术的具体实现与应用吧... 目录引言一、实现原理与基础方法1.1 核心函数解析1.2 基础实现模板1.3 整数版实现二、典型应用场景2

Python实现无痛修改第三方库源码的方法详解

《Python实现无痛修改第三方库源码的方法详解》很多时候,我们下载的第三方库是不会有需求不满足的情况,但也有极少的情况,第三方库没有兼顾到需求,本文将介绍几个修改源码的操作,大家可以根据需求进行选择... 目录需求不符合模拟示例 1. 修改源文件2. 继承修改3. 猴子补丁4. 追踪局部变量需求不符合很

Java的IO模型、Netty原理解析

《Java的IO模型、Netty原理解析》Java的I/O是以流的方式进行数据输入输出的,Java的类库涉及很多领域的IO内容:标准的输入输出,文件的操作、网络上的数据传输流、字符串流、对象流等,这篇... 目录1.什么是IO2.同步与异步、阻塞与非阻塞3.三种IO模型BIO(blocking I/O)NI

Spring事务中@Transactional注解不生效的原因分析与解决

《Spring事务中@Transactional注解不生效的原因分析与解决》在Spring框架中,@Transactional注解是管理数据库事务的核心方式,本文将深入分析事务自调用的底层原理,解释为... 目录1. 引言2. 事务自调用问题重现2.1 示例代码2.2 问题现象3. 为什么事务自调用会失效3

找不到Anaconda prompt终端的原因分析及解决方案

《找不到Anacondaprompt终端的原因分析及解决方案》因为anaconda还没有初始化,在安装anaconda的过程中,有一行是否要添加anaconda到菜单目录中,由于没有勾选,导致没有菜... 目录问题原因问http://www.chinasem.cn题解决安装了 Anaconda 却找不到 An

Spring定时任务只执行一次的原因分析与解决方案

《Spring定时任务只执行一次的原因分析与解决方案》在使用Spring的@Scheduled定时任务时,你是否遇到过任务只执行一次,后续不再触发的情况?这种情况可能由多种原因导致,如未启用调度、线程... 目录1. 问题背景2. Spring定时任务的基本用法3. 为什么定时任务只执行一次?3.1 未启用

C++ 各种map特点对比分析

《C++各种map特点对比分析》文章比较了C++中不同类型的map(如std::map,std::unordered_map,std::multimap,std::unordered_multima... 目录特点比较C++ 示例代码 ​​​​​​代码解释特点比较1. std::map底层实现:基于红黑

Spring、Spring Boot、Spring Cloud 的区别与联系分析

《Spring、SpringBoot、SpringCloud的区别与联系分析》Spring、SpringBoot和SpringCloud是Java开发中常用的框架,分别针对企业级应用开发、快速开... 目录1. Spring 框架2. Spring Boot3. Spring Cloud总结1. Sprin