为什么使用Java8中的并行流运算耗时变长了?

2024-06-11 11:20

本文主要是介绍为什么使用Java8中的并行流运算耗时变长了?,希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!

写在文章开头

近期对迭代的功能进行压测检查,发现某些使用并发技术的线程任务耗时非常漫长,结合监控排查定位到的并行流使用上的不恰当,遂以此文分享一下笔者发现的问题。

在这里插入图片描述

Hi,我是 sharkChili ,是个不断在硬核技术上作死的 java coder ,是 CSDN的博客专家 ,也是开源项目 Java Guide 的维护者之一,熟悉 Java 也会一点 Go ,偶尔也会在 C源码 边缘徘徊。写过很多有意思的技术博客,也还在研究并输出技术的路上,希望我的文章对你有帮助,非常欢迎你关注我的公众号: 写代码的SharkChili

因为近期收到很多读者的私信,所以也专门创建了一个交流群,感兴趣的读者可以通过上方的公众号获取笔者的联系方式完成好友添加,点击备注 “加群” 即可和笔者和笔者的朋友们进行深入交流。

在这里插入图片描述

问题复现

需求背景

这里笔者先简单介绍一下当前功能的使用背景,当前功能是一些大数据量的计算密集型任务定时执行,在常规优化效率有限的情况下,考虑到复用性,笔者通过JDK8底层内置的并行流完成这些任务的计算。

对应优化思路如下,可以看到针对每一批数据,笔者都是通过并行流采集出集合并将其写入文档:

在这里插入图片描述

常规串行计算

我们给出第一段代码示例,为了更专注于本文并行流问题的剖析,笔者对于两个并行线程所执行的数据采集和写入文档的操作通过原子类并发计算来模拟:

public static void main(String[] args) throws Exception {AtomicInteger atomicInteger = new AtomicInteger();CountDownLatch countDownLatch = new CountDownLatch(2);long beginTime = System.currentTimeMillis();//模拟采集5000w数据并写入本地文档中new Thread(() -> {IntStream.range(0, 5000_0000).forEach(i -> atomicInteger.getAndIncrement());countDownLatch.countDown();}, "t1").start();//模拟采集5000w数据并写入本地文档中new Thread(() -> {IntStream.range(0, 5000_0000).forEach(i -> atomicInteger.getAndIncrement());countDownLatch.countDown();}, "t2").start();//等待两个线程结束countDownLatch.await();//输出耗时long endTime = System.currentTimeMillis();System.out.println("atomicInteger: " + atomicInteger.get());System.out.println("time: " + (endTime - beginTime) + " ms");}

输出结果如下,可以看到1e的数据耗时大约需要1.6s

atomicInteger: 100000000
time: 1620 ms

单任务并行流

我们再进行更进一步的优化,将某个线程的任务使用并行流进行原子运算(模拟业务操作):

public static void main(String[] args) throws Exception {AtomicInteger atomicInteger = new AtomicInteger();CountDownLatch countDownLatch = new CountDownLatch(2);long beginTime = System.currentTimeMillis();//模拟并行流采集5000w数据并写入本地文档中new Thread(() -> {IntStream.range(0, 5000_0000).parallel().forEach(i -> atomicInteger.getAndIncrement());countDownLatch.countDown();}, "t1").start();//模拟采集5000w数据并写入本地文档中new Thread(() -> {IntStream.range(0, 5000_0000).forEach(i -> atomicInteger.getAndIncrement());countDownLatch.countDown();}, "t2").start();//等待两个线程结束countDownLatch.await();//输出耗时long endTime = System.currentTimeMillis();System.out.println("atomicInteger: " + atomicInteger.get());System.out.println("time: " + (endTime - beginTime) + " ms");}

从输出结果来看,性能表现提升了几毫秒,相对于最后生产上业务的数据量而言,可能会提升更多:

atomicInteger: 100000000
time: 1337 ms

双并行流运算

结合上述结果,我们大胆提出,是否所有任务都通过通过并行流进行运算,程序的执行性能是否会在此提升:

public static void main(String[] args) throws Exception {AtomicInteger atomicInteger = new AtomicInteger();CountDownLatch countDownLatch = new CountDownLatch(2);long beginTime = System.currentTimeMillis();//模拟并行流采集5000w数据并写入本地文档中new Thread(() -> {IntStream.range(0, 5000_0000).parallel().forEach(i -> atomicInteger.getAndIncrement());countDownLatch.countDown();}, "t1").start();//模拟并行流采集5000w数据并写入本地文档中new Thread(() -> {IntStream.range(0, 5000_0000).parallel().forEach(i -> atomicInteger.getAndIncrement());countDownLatch.countDown();}, "t2").start();//等待两个线程结束countDownLatch.await();//输出耗时long endTime = System.currentTimeMillis();System.out.println("atomicInteger: " + atomicInteger.get());System.out.println("time: " + (endTime - beginTime) + " ms");}

很明显,从最终的耗时来看,执行时间不减反增了,这是为什么呢?

atomicInteger: 100000000
time: 1863 ms

详解多任务采用并行流导致执行低效的原因

实际上并行流底层所采用的线程池是一个在程序启动初始化期间就会创建的线程池common,程序初始化时它会检查用户的是否有配置java.util.concurrent.ForkJoinPool.common.parallelism这个参数,如果有则基于这个参数的数值为common创建定量的线程,后续的我们的并行流运算的执行都会提交到该线程池中。

这就意味着我们上述的操作中,所有线程中千万的执行子项都通过同一个线程池进行并行运算,这期间线程池的忙碌程度可想而知,这也就是为什么笔者在进行压测时明明某些数据量不是很大的任务耗时却非常大的本质原因:

在这里插入图片描述

对于该问题,笔者也通过StackOverflow看到并行流设计的思想,设计者认为对于计算密集型任务,默认情况下,它将通过一个初始化一个CPU核心数一致的线程池,让所有并行运算共享一个线程池,进行并行流运算时使用的线程永远在核心数以内,由此也会出现相同的缺点,所有并行运算依赖同一个线程池,可能会导致大量任务大耗时或者大阻塞:

This also means if you have nested parallel streams or multiple parallel streams started concurrently, they will all share the same pool. Advantage: you will never use more than the default (number of available processors). Disadvantage: you may not get “all the processors” assigned to each parallel stream you initiate (if you happen to have more than one). (Apparently you can use a ManagedBlocker to circumvent that.)

这一点我们也可以在ForkJoinPool的静态代码块中

static {// initialize field offsets for CAS etctry {//......//调用makeCommonPool完成线程池创建和初始化common = java.security.AccessController.doPrivileged(new java.security.PrivilegedAction<ForkJoinPool>() {public ForkJoinPool run() { return makeCommonPool(); }});int par = common.config & SMASK; // report 1 even if threads disabledcommonParallelism = par > 0 ? par : 1;}

对应的我们步入makeCommonPool方法即可看到线程池的创建逻辑,即判断用户是否有通过java.util.concurrent.ForkJoinPool.common.parallelism指定线程数,若没有则按照CPU核心数完成初始化:

private static ForkJoinPool makeCommonPool() {//......try {  // ignore exceptions in accessing/parsing properties//获取用户对于common线程池中线程数的配置String pp = System.getProperty("java.util.concurrent.ForkJoinPool.common.parallelism");if (pp != null)parallelism = Integer.parseInt(pp);//......} catch (Exception ignore) {}//......//若小于parallelism小于0则说明用户没有指定,则直接按照CPU核心数创建线程池if (parallelism < 0 && // default 1 less than #cores(parallelism = Runtime.getRuntime().availableProcessors() - 1) <= 0)parallelism = 1;//基于CPU核心数创建 ForkJoinPool线程池return new ForkJoinPool(parallelism, factory, handler, LIFO_QUEUE,"ForkJoinPool.commonPool-worker-");}

解决方案

很明显,对于该问题就是因为多个并行运算跑到了单个线程池中,我们的解决方式无非是以下几种:

  1. 提升线程池线程数量已处理更多的并发运算。
  2. 业务上避免大量并发运算去竞争common线程池。

结合业务场景,笔者对于并行流的使用更多是计算密集型任务,通过java.util.concurrent.ForkJoinPool.common.parallelism去提升线程数并不会带来提升,所以在笔者结合业务场景通过压测计算出每个定时任务的耗时,大约是5分钟,所以笔者通过调整定时任务的执行间隔由原来的3min改为5min保证任务错峰执行解决该问题:

在这里插入图片描述

小结

我是 sharkchiliCSDN Java 领域博客专家开源项目—JavaGuide contributor,我想写一些有意思的东西,希望对你有帮助,如果你想实时收到我写的硬核的文章也欢迎你关注我的公众号: 写代码的SharkChili
因为近期收到很多读者的私信,所以也专门创建了一个交流群,感兴趣的读者可以通过上方的公众号获取笔者的联系方式完成好友添加,点击备注 “加群” 即可和笔者和笔者的朋友们进行深入交流。

在这里插入图片描述

参考

Custom thread pool in Java 8 parallel stream:https://stackoverflow.com/questions/21163108/custom-thread-pool-in-java-8-parallel-stream

ForkJoinPool的commonPool相关参数配置
:https://www.jianshu.com/p/1b5f4ea0074a

这篇关于为什么使用Java8中的并行流运算耗时变长了?的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!



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

相关文章

JVM 的类初始化机制

前言 当你在 Java 程序中new对象时,有没有考虑过 JVM 是如何把静态的字节码(byte code)转化为运行时对象的呢,这个问题看似简单,但清楚的同学相信也不会太多,这篇文章首先介绍 JVM 类初始化的机制,然后给出几个易出错的实例来分析,帮助大家更好理解这个知识点。 JVM 将字节码转化为运行时对象分为三个阶段,分别是:loading 、Linking、initialization

Spring Security 基于表达式的权限控制

前言 spring security 3.0已经可以使用spring el表达式来控制授权,允许在表达式中使用复杂的布尔逻辑来控制访问的权限。 常见的表达式 Spring Security可用表达式对象的基类是SecurityExpressionRoot。 表达式描述hasRole([role])用户拥有制定的角色时返回true (Spring security默认会带有ROLE_前缀),去

浅析Spring Security认证过程

类图 为了方便理解Spring Security认证流程,特意画了如下的类图,包含相关的核心认证类 概述 核心验证器 AuthenticationManager 该对象提供了认证方法的入口,接收一个Authentiaton对象作为参数; public interface AuthenticationManager {Authentication authenticate(Authenti

Spring Security--Architecture Overview

1 核心组件 这一节主要介绍一些在Spring Security中常见且核心的Java类,它们之间的依赖,构建起了整个框架。想要理解整个架构,最起码得对这些类眼熟。 1.1 SecurityContextHolder SecurityContextHolder用于存储安全上下文(security context)的信息。当前操作的用户是谁,该用户是否已经被认证,他拥有哪些角色权限…这些都被保

Spring Security基于数据库验证流程详解

Spring Security 校验流程图 相关解释说明(认真看哦) AbstractAuthenticationProcessingFilter 抽象类 /*** 调用 #requiresAuthentication(HttpServletRequest, HttpServletResponse) 决定是否需要进行验证操作。* 如果需要验证,则会调用 #attemptAuthentica

Spring Security 从入门到进阶系列教程

Spring Security 入门系列 《保护 Web 应用的安全》 《Spring-Security-入门(一):登录与退出》 《Spring-Security-入门(二):基于数据库验证》 《Spring-Security-入门(三):密码加密》 《Spring-Security-入门(四):自定义-Filter》 《Spring-Security-入门(五):在 Sprin

Java架构师知识体认识

源码分析 常用设计模式 Proxy代理模式Factory工厂模式Singleton单例模式Delegate委派模式Strategy策略模式Prototype原型模式Template模板模式 Spring5 beans 接口实例化代理Bean操作 Context Ioc容器设计原理及高级特性Aop设计原理Factorybean与Beanfactory Transaction 声明式事物

中文分词jieba库的使用与实景应用(一)

知识星球:https://articles.zsxq.com/id_fxvgc803qmr2.html 目录 一.定义: 精确模式(默认模式): 全模式: 搜索引擎模式: paddle 模式(基于深度学习的分词模式): 二 自定义词典 三.文本解析   调整词出现的频率 四. 关键词提取 A. 基于TF-IDF算法的关键词提取 B. 基于TextRank算法的关键词提取

使用SecondaryNameNode恢复NameNode的数据

1)需求: NameNode进程挂了并且存储的数据也丢失了,如何恢复NameNode 此种方式恢复的数据可能存在小部分数据的丢失。 2)故障模拟 (1)kill -9 NameNode进程 [lytfly@hadoop102 current]$ kill -9 19886 (2)删除NameNode存储的数据(/opt/module/hadoop-3.1.4/data/tmp/dfs/na

Hadoop数据压缩使用介绍

一、压缩原则 (1)运算密集型的Job,少用压缩 (2)IO密集型的Job,多用压缩 二、压缩算法比较 三、压缩位置选择 四、压缩参数配置 1)为了支持多种压缩/解压缩算法,Hadoop引入了编码/解码器 2)要在Hadoop中启用压缩,可以配置如下参数