深读源码-java线程系列之ForkJoinPool深入解析

2023-11-07 09:18

本文主要是介绍深读源码-java线程系列之ForkJoinPool深入解析,希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!

简介

随着在硬件上多核处理器的发展和广泛使用,并发编程成为程序员必须掌握的一门技术,在面试中也经常考查面试者并发相关的知识。

今天,我们就来看一道面试题:

如何充分利用多核CPU,计算很大数组中所有整数的和?

剖析

  • 单线程相加?

我们最容易想到就是单线程相加,一个for循环搞定。

  • 线程池相加?

如果进一步优化,我们会自然而然地想到使用线程池来分段相加,最后再把每个段的结果相加。

  • 其它?

Yes,就是我们今天的主角——ForkJoinPool,但是它要怎么实现呢?似乎没怎么用过哈^^

三种实现

OK,剖析完了,我们直接来看三种实现,不墨迹,直接上菜。

/*** 计算1亿个整数的和*/
public class ForkJoinPoolTest01 {public static void main(String[] args) throws ExecutionException, InterruptedException {// 构造数据int length = 100000000;long[] arr = new long[length];for (int i = 0; i < length; i++) {arr[i] = ThreadLocalRandom.current().nextInt(Integer.MAX_VALUE);}// 单线程singleThreadSum(arr);// ThreadPoolExecutor线程池multiThreadSum(arr);// ForkJoinPool线程池forkJoinSum(arr);}private static void singleThreadSum(long[] arr) {long start = System.currentTimeMillis();long sum = 0;for (int i = 0; i < arr.length; i++) {// 模拟耗时sum += (arr[i]/3*3/3*3/3*3/3*3/3*3);}System.out.println("sum: " + sum);System.out.println("single thread elapse: " + (System.currentTimeMillis() - start));}private static void multiThreadSum(long[] arr) throws ExecutionException, InterruptedException {long start = System.currentTimeMillis();int count = 8;ExecutorService threadPool = Executors.newFixedThreadPool(count);List<Future<Long>> list = new ArrayList<>();for (int i = 0; i < count; i++) {int num = i;// 分段提交任务Future<Long> future = threadPool.submit(() -> {long sum = 0;for (int j = arr.length / count * num; j < (arr.length / count * (num + 1)); j++) {try {// 模拟耗时sum += (arr[j]/3*3/3*3/3*3/3*3/3*3);} catch (Exception e) {e.printStackTrace();}}return sum;});list.add(future);}// 每个段结果相加long sum = 0;for (Future<Long> future : list) {sum += future.get();}System.out.println("sum: " + sum);System.out.println("multi thread elapse: " + (System.currentTimeMillis() - start));}private static void forkJoinSum(long[] arr) throws ExecutionException, InterruptedException {long start = System.currentTimeMillis();ForkJoinPool forkJoinPool = ForkJoinPool.commonPool();// 提交任务ForkJoinTask<Long> forkJoinTask = forkJoinPool.submit(new SumTask(arr, 0, arr.length));// 获取结果Long sum = forkJoinTask.get();forkJoinPool.shutdown();System.out.println("sum: " + sum);System.out.println("fork join elapse: " + (System.currentTimeMillis() - start));}private static class SumTask extends RecursiveTask<Long> {private long[] arr;private int from;private int to;public SumTask(long[] arr, int from, int to) {this.arr = arr;this.from = from;this.to = to;}@Overrideprotected Long compute() {// 小于1000的时候直接相加,可灵活调整if (to - from <= 1000) {long sum = 0;for (int i = from; i < to; i++) {// 模拟耗时sum += (arr[i]/3*3/3*3/3*3/3*3/3*3);}return sum;}// 分成两段任务int middle = (from + to) / 2;SumTask left = new SumTask(arr, from, middle);SumTask right = new SumTask(arr, middle, to);// 提交左边的任务left.fork();// 右边的任务直接利用当前线程计算,节约开销Long rightResult = right.compute();// 等待左边计算完毕Long leftResult = left.join();// 返回结果return leftResult + rightResult;}}
}

实际上计算1亿个整数相加,单线程是最快的,我的电脑大概是100ms左右,使用线程池反而会变慢。

所以,为了演示ForkJoinPool的牛逼之处,我把每个数都/3*3/3*3/3*3/3*3/3*3了一顿操作,用来模拟计算耗时。

来看结果:

sum: 107352457433800662
single thread elapse: 789
sum: 107352457433800662
multi thread elapse: 228
sum: 107352457433800662
fork join elapse: 189

可以看到,ForkJoinPool相对普通线程池还是有很大提升的。

问题:普通线程池能否实现ForkJoinPool这种计算方式呢,即大任务拆中任务,中任务拆小任务,最后再汇总?

 

你可以试试看(-᷅_-᷄)

OK,下面我们正式进入ForkJoinPool的解析。

分治法

  • 基本思想

把一个规模大的问题划分为规模较小的子问题,然后分而治之,最后合并子问题的解得到原问题的解。

  • 步骤

(1)分割原问题:

(2)求解子问题:

(3)合并子问题的解为原问题的解。

在分治法中,子问题一般是相互独立的,因此,经常通过递归调用算法来求解子问题。

  • 典型应用场景

(1)二分搜索

(2)大整数乘法

(3)Strassen矩阵乘法

(4)棋盘覆盖

(5)归并排序

(6)快速排序

(7)线性时间选择

(8)汉诺塔

ForkJoinPool继承体系

ForkJoinPool是 java 7 中新增的线程池类,它的继承体系如下:

 

ForkJoinPool和ThreadPoolExecutor都是继承自AbstractExecutorService抽象类,所以它和ThreadPoolExecutor的使用几乎没有多少区别,除了任务变成了ForkJoinTask以外。

这里又运用到了一种很重要的设计原则——开闭原则——对修改关闭,对扩展开放。

可见整个线程池体系一开始的接口设计就很好,新增一个线程池类,不会对原有的代码造成干扰,还能利用原有的特性。

ForkJoinTask

两个主要方法

  • fork()

fork()方法类似于线程的Thread.start()方法,但是它不是真的启动一个线程,而是将任务放入到工作队列中。

  • join()

join()方法类似于线程的Thread.join()方法,但是它不是简单地阻塞线程,而是利用工作线程运行其它任务。当一个工作线程中调用了join()方法,它将处理其它任务,直到注意到目标子任务已经完成了。

三个子类

  • RecursiveAction

无返回值任务。

  • RecursiveTask

有返回值任务。

  • CountedCompleter

无返回值任务,完成任务后可以触发回调。

ForkJoinPool内部原理

ForkJoinPool内部使用的是“工作窃取”算法实现的。

forkjoinpool

(1)每个工作线程都有自己的工作队列WorkQueue;

(2)这是一个双端队列,它是线程私有的;

(3)ForkJoinTask中fork的子任务,将放入运行该任务的工作线程的队头,工作线程将以LIFO的顺序来处理工作队列中的任务;

(4)为了最大化地利用CPU,空闲的线程将从其它线程的队列中“窃取”任务来执行;

(5)从工作队列的尾部窃取任务,以减少竞争;

(6)双端队列的操作:push()/pop()仅在其所有者工作线程中调用,poll()是由其它线程窃取任务时调用的;

(7)当只剩下最后一个任务时,还是会存在竞争,是通过CAS来实现的;

forkjoinpool

ForkJoinPool最佳实践

(1)最适合的是计算密集型任务;

(2)在需要阻塞工作线程时,可以使用ManagedBlocker;

(3)不应该在RecursiveTask的内部使用ForkJoinPool.invoke()/invokeAll();

总结

(1)ForkJoinPool特别适合于“分而治之”算法的实现;

(2)ForkJoinPool和ThreadPoolExecutor是互补的,不是谁替代谁的关系,二者适用的场景不同;

(3)ForkJoinTask有两个核心方法——fork()和join(),有三个重要子类——RecursiveAction、RecursiveTask和CountedCompleter;

(4)ForkjoinPool内部基于“工作窃取”算法实现;

(5)每个线程有自己的工作队列,它是一个双端队列,自己从队列头存取任务,其它线程从尾部窃取任务;

(6)ForkJoinPool最适合于计算密集型任务,但也可以使用ManagedBlocker以便用于阻塞型任务;

(7)RecursiveTask内部可以少调用一次fork(),利用当前线程处理,这是一种技巧;

彩蛋

ManagedBlocker怎么使用?

答:ManagedBlocker相当于明确告诉ForkJoinPool框架要阻塞了,ForkJoinPool就会启另一个线程来运行任务,以最大化地利用CPU。

请看下面的例子,自己琢磨哈^^。

/*** 斐波那契数列* 一个数是它前面两个数之和* 1,1,2,3,5,8,13,21*/
public class Fibonacci {public static void main(String[] args) {long time = System.currentTimeMillis();Fibonacci fib = new Fibonacci();int result = fib.f(1_000).bitCount();time = System.currentTimeMillis() - time;System.out.println("result= " + result);System.out.println("test1_000() time = " + time);}public BigInteger f(int n) {Map<Integer, BigInteger> cache = new ConcurrentHashMap<>();cache.put(0, BigInteger.ZERO);cache.put(1, BigInteger.ONE);return f(n, cache);}private final BigInteger RESERVED = BigInteger.valueOf(-1000);public BigInteger f(int n, Map<Integer, BigInteger> cache) {BigInteger result = cache.putIfAbsent(n, RESERVED);if (result == null) {int half = (n + 1) / 2;RecursiveTask<BigInteger> f0_task = new RecursiveTask<BigInteger>() {@Overrideprotected BigInteger compute() {return f(half - 1, cache);}};f0_task.fork();BigInteger f1 = f(half, cache);BigInteger f0 = f0_task.join();long time = n > 10_000 ? System.currentTimeMillis() : 0;try {if (n % 2 == 1) {result = f0.multiply(f0).add(f1.multiply(f1));} else {result = f0.shiftLeft(1).add(f1).multiply(f1);}synchronized (RESERVED) {cache.put(n, result);RESERVED.notifyAll();}} finally {time = n > 10_000 ? System.currentTimeMillis() - time : 0;if (time > 50)System.out.printf("f(%d) took %d%n", n, time);}} else if (result == RESERVED) {try {ReservedFibonacciBlocker blocker = new ReservedFibonacciBlocker(n, cache);ForkJoinPool.managedBlock(blocker);result = blocker.result;} catch (InterruptedException e) {throw new CancellationException("interrupted");}}return result;// return f(n - 1).add(f(n - 2));}private class ReservedFibonacciBlocker implements ForkJoinPool.ManagedBlocker {private BigInteger result;private final int n;private final Map<Integer, BigInteger> cache;public ReservedFibonacciBlocker(int n, Map<Integer, BigInteger> cache) {this.n = n;this.cache = cache;}@Overridepublic boolean block() throws InterruptedException {synchronized (RESERVED) {while (!isReleasable()) {RESERVED.wait();}}return true;}@Overridepublic boolean isReleasable() {return (result = cache.get(n)) != RESERVED;}}
}

原文链接:https://www.cnblogs.com/tong-yuan/p/11824018.html

这篇关于深读源码-java线程系列之ForkJoinPool深入解析的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!



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

相关文章

Java中Arrays类和Collections类常用方法示例详解

《Java中Arrays类和Collections类常用方法示例详解》本文总结了Java中Arrays和Collections类的常用方法,涵盖数组填充、排序、搜索、复制、列表转换等操作,帮助开发者高... 目录Arrays.fill()相关用法Arrays.toString()Arrays.sort()A

Spring Boot Maven 插件如何构建可执行 JAR 的核心配置

《SpringBootMaven插件如何构建可执行JAR的核心配置》SpringBoot核心Maven插件,用于生成可执行JAR/WAR,内置服务器简化部署,支持热部署、多环境配置及依赖管理... 目录前言一、插件的核心功能与目标1.1 插件的定位1.2 插件的 Goals(目标)1.3 插件定位1.4 核

如何使用Lombok进行spring 注入

《如何使用Lombok进行spring注入》本文介绍如何用Lombok简化Spring注入,推荐优先使用setter注入,通过注解自动生成getter/setter及构造器,减少冗余代码,提升开发效... Lombok为了开发环境简化代码,好处不用多说。spring 注入方式为2种,构造器注入和setter

使用zip4j实现Java中的ZIP文件加密压缩的操作方法

《使用zip4j实现Java中的ZIP文件加密压缩的操作方法》本文介绍如何通过Maven集成zip4j1.3.2库创建带密码保护的ZIP文件,涵盖依赖配置、代码示例及加密原理,确保数据安全性,感兴趣的... 目录1. zip4j库介绍和版本1.1 zip4j库概述1.2 zip4j的版本演变1.3 zip4

Java堆转储文件之1.6G大文件处理完整指南

《Java堆转储文件之1.6G大文件处理完整指南》堆转储文件是优化、分析内存消耗的重要工具,:本文主要介绍Java堆转储文件之1.6G大文件处理的相关资料,文中通过代码介绍的非常详细,需要的朋友可... 目录前言文件为什么这么大?如何处理这个文件?分析文件内容(推荐)删除文件(如果不需要)查看错误来源如何避

SpringBoot整合Dubbo+ZK注册失败的坑及解决

《SpringBoot整合Dubbo+ZK注册失败的坑及解决》使用Dubbo框架时,需在公共pom添加依赖,启动类加@EnableDubbo,实现类用@DubboService替代@Service,配... 目录1.先看下公共的pom(maven创建的pom工程)2.启动类上加@EnableDubbo3.实

SpringBoot整合(ES)ElasticSearch7.8实践

《SpringBoot整合(ES)ElasticSearch7.8实践》本文详细介绍了SpringBoot整合ElasticSearch7.8的教程,涵盖依赖添加、客户端初始化、索引创建与获取、批量插... 目录SpringBoot整合ElasticSearch7.8添加依赖初始化创建SpringBoot项

JAVA覆盖和重写的区别及说明

《JAVA覆盖和重写的区别及说明》非静态方法的覆盖即重写,具有多态性;静态方法无法被覆盖,但可被重写(仅通过类名调用),二者区别在于绑定时机与引用类型关联性... 目录Java覆盖和重写的区别经常听到两种话认真读完上面两份代码JAVA覆盖和重写的区别经常听到两种话1.覆盖=重写。2.静态方法可andro

SpringBoot中六种批量更新Mysql的方式效率对比分析

《SpringBoot中六种批量更新Mysql的方式效率对比分析》文章比较了MySQL大数据量批量更新的多种方法,指出REPLACEINTO和ONDUPLICATEKEY效率最高但存在数据风险,MyB... 目录效率比较测试结构数据库初始化测试数据批量修改方案第一种 for第二种 case when第三种

Java docx4j高效处理Word文档的实战指南

《Javadocx4j高效处理Word文档的实战指南》对于需要在Java应用程序中生成、修改或处理Word文档的开发者来说,docx4j是一个强大而专业的选择,下面我们就来看看docx4j的具体使用... 目录引言一、环境准备与基础配置1.1 Maven依赖配置1.2 初始化测试类二、增强版文档操作示例2.