【揭秘】ForkJoinPool全面解析

简介: ForkJoinPool是Java中的并行计算框架,其优点在于能够高效利用多核处理器资源,它采用分治策略将大任务拆分成小任务,通过工作窃取算法平衡负载,从而实现任务的并行执行和快速完成,此外,ForkJoinPool还提供了简洁的API和丰富的任务控制机制,支撑开发人员开发高效的并行代码。

【揭秘】ForkJoinPool全面解析 - 程序员古德

文章摘要

ForkJoinPool是Java中的并行计算框架,其优点在于能够高效利用多核处理器资源,它采用分治策略将大任务拆分成小任务,通过工作窃取算法平衡负载,从而实现任务的并行执行和快速完成,此外,ForkJoinPool还提供了简洁的API和丰富的任务控制机制,支撑开发人员开发高效的并行代码。

核心概念

ForkJoinPool 是 Java 并发包 java.util.concurrent 中的一个类,它主要用于解决可以通过分治策略(Divide-and-Conquer)来并行处理的问题,这类问题通常可以被分解为更小的子问题,子问题和原问题在结构上相同或类似,只不过规模不同,通过递归地将问题分解为更小的部分,ForkJoinPool 可以利用多核处理器并行地处理这些子问题,然后再将结果合并起来,从而高效地解决问题。

ForkJoinPool 的主要特点包括:

  1. 工作窃取算法(Work-Stealing Algorithm):当一个线程完成了自己的任务后,它可以从其他线程的任务队列中“窃取”任务来执行,这有助于平衡负载和提高处理器的利用率。
  2. 递归分解与合并:非常适合处理可以递归分解的问题,如排序、搜索、数值计算等,开发者需要实现 ForkJoinTask 接口(通常使用它的子类 RecursiveAction 用于无返回值的任务,或使用 RecursiveTask 用于有返回值的任务)来定义问题的分解和结果的合并。
  3. 非阻塞设计:使用内部队列来管理任务,避免了使用锁或其他同步机制,从而减少了线程间的竞争和阻塞。
  4. 并行度控制:允许开发者控制并行执行的线程数量,可以根据处理器的核心数来优化性能。

ForkJoinPool 适用于那些可以自然分解为多个独立子任务,并且这些子任务之间不需要太多通信或同步的问题,常见的使用场景包括并行数组处理(如排序、过滤、映射)、并行集合处理(如归约操作)、科学计算中的并行算法(如矩阵乘法、快速傅里叶变换)等。

代码案例

import java.util.concurrent.ForkJoinPool;  
import java.util.concurrent.RecursiveAction;  

public class ForkJoinSumCalculator {
   
     

    public static void main(String[] args) {
   
     
        // 定义一个需要求和的数组  
        int[] numbers = {
   
   1, 2, 3, 4, 5, 6, 7, 8, 9, 10};  

        // 创建一个ForkJoinPool实例,它将使用可用的所有处理器  
        ForkJoinPool pool = new ForkJoinPool();  

        // 创建一个ForkJoinTask来执行求和操作  
        SumTask task = new SumTask(numbers, 0, numbers.length);  

        // 提交任务到ForkJoinPool并等待它的完成  
        pool.invoke(task);  

        // 输出最终求和结果  
        System.out.println("Sum of all numbers: " + task.getSum());  

        // 关闭ForkJoinPool(虽然在这个例子中它并不是严格必要的,因为程序即将退出)  
        pool.shutdown();  
    }  

    // 定义一个继承自RecursiveAction的任务类  
    static class SumTask extends RecursiveAction {
   
     
        private static final long serialVersionUID = 1L;  

        // 阈值,当数组长度小于这个值时,直接计算结果而不再拆分  
        private static final int THRESHOLD = 5;  

        private int[] numbers;  
        private int startIndex;  
        private int endIndex;  
        private int sum; // 存储子数组的和  

        public SumTask(int[] numbers, int startIndex, int endIndex) {
   
     
            this.numbers = numbers;  
            this.startIndex = startIndex;  
            this.endIndex = endIndex;  
        }  

        // 获取当前任务计算的和  
        public int getSum() {
   
     
            return sum;  
        }  

        @Override  
        protected void compute() {
   
     
            // 如果任务足够小,直接计算  
            if (endIndex - startIndex <= THRESHOLD) {
   
     
                sum = calculateDirectly();  
            } else {
   
     
                // 否则,拆分任务  
                int middleIndex = startIndex + (endIndex - startIndex) / 2;  
                SumTask leftTask = new SumTask(numbers, startIndex, middleIndex);  
                SumTask rightTask = new SumTask(numbers, middleIndex, endIndex);  

                // 递归执行任务  
                invokeAll(leftTask, rightTask);  

                // 合并结果  
                sum = leftTask.getSum() + rightTask.getSum();  
            }  
        }  

        // 直接计算子数组的和  
        private int calculateDirectly() {
   
     
            int localSum = 0;  
            for (int i = startIndex; i < endIndex; i++) {
   
     
                localSum += numbers[i];  
            }  
            return localSum;  
        }  
    }  
}

在上面代码中,SumTask类有一个sum字段来存储计算的和,以及一个getSum方法来检索它,在compute方法中,如果任务的大小超过阈值,任务将被拆分为两个子任务,并且递归地执行,然后,将子任务的结果合并以计算总和,如果任务的大小小于或等于阈值,将直接计算子数组的和。

核心API

ForkJoinPool 提供了一个框架,用于将大任务分解成小任务,然后并行地执行这些小任务,最后再将结果合并起来,它提供的方法主要涉及到任务的提交、执行、管理和配置等方面,下面是一些常用方法的简要说明。

构造方法

  1. ForkJoinPool(): 创建一个默认并行级别的 ForkJoinPool,通常使用可用的处理器数量作为并行级别。
  2. ForkJoinPool(int parallelism): 创建一个具有指定并行级别的 ForkJoinPool

任务提交

  1. invoke(ForkJoinTask<?> task): 同步执行指定的任务,并等待其完成。

  2. submit(ForkJoinTask<?> task): 异步提交一个任务以供执行,并返回一个表示该任务的 Future

  3. execute(ForkJoinTask<?> task): 安排一个任务的执行,但不等待其完成。

任务管理

  1. awaitQuiescence(long timeout, TimeUnit unit): 等待所有任务完成执行,或者直到超时。
  2. shutdown(): 可能启动有序关闭,在该过程中执行现有任务但不接受新任务。
  3. shutdownNow(): 试图停止所有正在执行的活动任务,暂停处理正在等待的任务,并返回等待执行的任务列表。
  4. isShutdown(): 如果此池已关闭或正在关闭,则返回 true
  5. isTerminated(): 如果关闭后所有任务都已完成,则返回 true
  6. awaitTermination(long timeout, TimeUnit unit): 请求关闭并等待所有任务完成执行,或者直到超时。

获取任务结果

  1. 对于 RecursiveTask(有返回值的任务),通常会在调用任务的 join 方法时获取任务结果。

配置和状态

  1. getParallelism(): 返回此 ForkJoinPool 的并行级别。
  2. getPoolSize(): 返回此 ForkJoinPool 中的活动线程估计数。
  3. getActiveThreadCount(): 返回此 ForkJoinPool 中当前活动的线程数。
  4. getRunningThreadCount(): 返回此 ForkJoinPool 中正在运行任务的线程数。
  5. getQueuedTaskCount(): 返回此 ForkJoinPool 工作队列中待处理的任务数估计值。
  6. getStealCount(): 返回从此 ForkJoinPool 中成功窃取的任务数估计值。

核心总结

【揭秘】ForkJoinPool全面解析 - 程序员古德

ForkJoinPool是Java并行计算的利器,其优点在于能高效地将大任务拆成小任务,通过工作窃取机制充分利用多核处理器,加速任务执行,但它也有缺点,比如任务划分不均可能导致部分处理器闲置,且更适合计算密集型而非IO密集型任务,使用时,建议合理划分任务,保持任务均衡,同时注意异常处理和线程资源管理。

关注我,每天学习互联网编程技术 - 程序员古德

相关文章
|
存储 缓存 安全
高并发编程之阻塞队列
高并发编程之阻塞队列
178 1
|
JavaScript Java 大数据
分享Fork/Join经典案例
`shigen`是位专注于Java、Python、Vue和Shell的博主,分享技术成长和认知。上篇文章探讨了Java的Fork/Join框架,它类似线程池,通过拆分大任务并并行执行提升效率。以大序列求和为例展示了ForkJoinPool的使用,与普通线程池对比,Fork/Join效率提升约50%。适合递归任务、独立子任务和长执行时间的任务。注意任务粒度、避免共享状态和死锁。推荐观看相关视频深入理解。一起学习,每天进步!
151 0
分享Fork/Join经典案例
|
7月前
|
消息中间件 缓存 监控
MQ消息积压 / Rocketmq 积压 最全的处理方案。 (秒懂+图解+史上最全)
MQ消息积压 / Rocketmq 积压 最全的处理方案。 (秒懂+图解+史上最全)
MQ消息积压 / Rocketmq 积压 最全的处理方案。 (秒懂+图解+史上最全)
|
7月前
|
Java Spring
Spring Boot配置的优先级?
在Spring Boot项目中,配置可通过配置文件和外部配置实现。支持的配置文件包括application.properties、application.yml和application.yaml,优先级依次降低。外部配置常用方式有Java系统属性(如-Dserver.port=9001)和命令行参数(如--server.port=10010),其中命令行参数优先级高于系统属性。整体优先级顺序为:命令行参数 &gt; Java系统属性 &gt; application.properties &gt; application.yml &gt; application.yaml。
1158 0
|
消息中间件 存储 RocketMQ
Rocketmq如何保证消息不丢失
文章分析了RocketMQ如何通过生产者端的同步发送与重试机制、Broker端的持久化存储与消息重试投递策略、以及消费者端的手动提交ack与幂等性处理,来确保消息在整个传输和消费过程中的不丢失。
|
缓存 NoSQL Java
什么是缓存?如何在 Spring Boot 中使用缓存框架
什么是缓存?如何在 Spring Boot 中使用缓存框架
921 0
|
10月前
|
存储 安全 Java
Spring Security 入门与详解
Spring Security 是 Spring 框架中的核心安全模块,提供认证、授权及防护功能。本文详解其核心概念,包括认证(Authentication)、授权(Authorization)和过滤器链(Security Filter Chain)。同时,通过代码示例介绍基本配置,如 PasswordEncoder、UserDetailsService 和自定义登录页面等。最后总结常见问题与解决方法,助你快速掌握 Spring Security 的使用与优化。
2554 0
|
SQL 关系型数据库 MySQL
|
XML JSON API
[1688一件代发]API接口关键词搜索(1688.item_search)
1688 一件代发的关键词搜索 API 接口,可快速从平台海量商品库中查找相关信息。主要参数包括:q(搜索关键字)、start_price 和 end_price(价格区间)、page(页码)、page_size(每页显示数量)、sort(排序方式)及 filter(额外过滤条件)。此接口适用于需要精准定位商品的开发者和商家。
|
消息中间件 容灾 物联网
【RocketMQ系列十四】RocketMQ中消息堆积如何处理
【RocketMQ系列十四】RocketMQ中消息堆积如何处理
2901 3