文档章节

多线程 ForkJoinPool

枯藤KT
 枯藤KT
发布于 2018/12/06 18:50
字数 2021
阅读 3
收藏 0

阅读目录

背景:ForkJoinPool的优势在于,可以充分利用多cpu,多核cpu的优势,把一个任务拆分成多个“小任务”,把多个“小任务”放到多个处理器核心上并行执行;当多个“小任务”执行完成之后,再将这些执行结果合并起来即可。这种思想值得学习。

主要参考《疯狂java讲义》

使用

Java7 提供了ForkJoinPool来支持将一个任务拆分成多个“小任务”并行计算,再把多个“小任务”的结果合并成总的计算结果。

ForkJoinPool是ExecutorService的实现类,因此是一种特殊的线程池。

使用方法:创建了ForkJoinPool实例之后,就可以调用ForkJoinPool的submit(ForkJoinTask<T> task) 或invoke(ForkJoinTask<T> task)方法来执行指定任务了。

其中ForkJoinTask代表一个可以并行、合并的任务。ForkJoinTask是一个抽象类,它还有两个抽象子类:RecusiveAction和RecusiveTask。其中RecusiveTask代表有返回值的任务,而RecusiveAction代表没有返回值的任务。

下面的UML类图显示了ForkJoinPool、ForkJoinTask之间的关系:

 

举例

以还行没有返回值的“大任务”(简单低打印1~300的数值)为例,程序将一个“大任务”拆分成多个“小任务”,并将任务交给ForkJoinPool来执行

复制代码
/**
 * Project Name:Spring0725
 * File Name:ForkJoinPoolAction.java
 * Package Name:work1201.basic
 * Date:2017年12月4日下午2:26:55
 * Copyright (c) 2017, 深圳金融电子结算中心 All Rights Reserved.
 *
*/

package work1201.basic;

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

/**
 * ClassName:ForkJoinPoolAction <br/>
 * Function: 使用ForkJoinPool完成一个任务的分段执行
 * 简单的打印0-300的数值。用多线程实现并行执行
 * Date:     2017年12月4日 下午2:26:55 <br/>
 * @author   prd-lxw
 * @version   1.0
 * @since    JDK 1.7
 * @see      
 */
public class ForkJoinPoolAction {
    
    public static void main(String[] args) throws Exception{
        PrintTask task = new PrintTask(0, 300);
        //创建实例,并执行分割任务
        ForkJoinPool pool = new ForkJoinPool();
        pool.submit(task);
         //线程阻塞,等待所有任务完成
        pool.awaitTermination(2, TimeUnit.SECONDS);
        pool.shutdown();
    }
}

/**
 * ClassName: PrintTask <br/>
 * Function: 继承RecursiveAction来实现“可分解”的任务。
 * date: 2017年12月4日 下午5:17:41 <br/>
 *
 * @author prd-lxw
 * @version 1.0
 * @since JDK 1.7
 */
class PrintTask extends RecursiveAction{
    private static final int THRESHOLD = 50; //最多只能打印50个数
    private int start;
    private int end;
    
    

    public PrintTask(int start, int end) {
        super();
        this.start = start;
        this.end = end;
    }



    @Override
    protected void compute() {
        
        if(end - start < THRESHOLD){
            for(int i=start;i<end;i++){
                System.out.println(Thread.currentThread().getName()+"的i值:"+i);
            }
        }else {
            int middle =(start+end)/2;
            PrintTask left = new PrintTask(start, middle);
            PrintTask right = new PrintTask(middle, end);
            //并行执行两个“小任务”
            left.fork();
            right.fork();
        }
        
    }
    
}
复制代码

 

执行结果:

复制代码
ForkJoinPool-1-worker-1的i值:262
ForkJoinPool-1-worker-7的i值:75
ForkJoinPool-1-worker-7的i值:76
ForkJoinPool-1-worker-5的i值:225
ForkJoinPool-1-worker-3的i值:187
ForkJoinPool-1-worker-6的i值:150
ForkJoinPool-1-worker-6的i值:151
ForkJoinPool-1-worker-6的i值:152
ForkJoinPool-1-worker-6的i值:153
ForkJoinPool-1-worker-6的i值:154
......
复制代码

 

因为我的电脑是i7处理器,一共8个cpu,观察线程的名称可以发现,8个cpu都在运行。

通过RecursiveTask的返回值,来对一个长度为100的数组元素进行累加。

复制代码
/**
 * Project Name:Spring0725
 * File Name:ForJoinPollTask.java
 * Package Name:work1201.basic
 * Date:2017年12月4日下午5:41:46
 * Copyright (c) 2017, 深圳金融电子结算中心 All Rights Reserved.
 *
*/

package work1201.basic;

import java.util.Random;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.Future;
import java.util.concurrent.RecursiveTask;

/**
 * ClassName:ForJoinPollTask <br/>
 * Function: 对一个长度为100的元素值进行累加
 * Date:     2017年12月4日 下午5:41:46 <br/>
 * @author   prd-lxw
 * @version   1.0
 * @since    JDK 1.7
 * @see      
 */
public class ForJoinPollTask {

    public static void main(String[] args) throws Exception {
        int[] arr = new int[100];
        Random random = new Random();
        int total =0;
        //初始化100个数组元素
        for(int i=0,len = arr.length;i<len;i++){
            int temp = random.nextInt(20);
            //对数组元素赋值,并将数组元素的值添加到sum总和中
            total += (arr[i]=temp);
        }
        System.out.println("初始化数组总和:"+total);
        SumTask task = new SumTask(arr, 0, arr.length);
//        创建一个通用池,这个是jdk1.8提供的功能
        ForkJoinPool pool = ForkJoinPool.commonPool();
        Future<Integer> future = pool.submit(task); //提交分解的SumTask 任务
        System.out.println("多线程执行结果:"+future.get());
        pool.shutdown(); //关闭线程池
        
        

    }

}

/**
 * ClassName: SumTask <br/>
 * Function: 继承抽象类RecursiveTask,通过返回的结果,来实现数组的多线程分段累累加
 *  RecursiveTask 具有返回值
 * date: 2017年12月4日 下午6:08:11 <br/>
 *
 * @author prd-lxw
 * @version 1.0
 * @since JDK 1.7
 */
class SumTask extends RecursiveTask<Integer>{
    private static final int THRESHOLD = 20; //每个小任务 最多只累加20个数
    private int arry[];
    private int start;
    private int end;
    
    

    /**
     * Creates a new instance of SumTask.
     * 累加从start到end的arry数组
     * @param arry
     * @param start
     * @param end
     */
    public SumTask(int[] arry, int start, int end) {
        super();
        this.arry = arry;
        this.start = start;
        this.end = end;
    }



    @Override
    protected Integer compute() {
        int sum =0;
        //当end与start之间的差小于threshold时,开始进行实际的累加
        if(end - start <THRESHOLD){
            for(int i= start;i<end;i++){
                sum += arry[i];
            }
            return sum;
        }else {//当end与start之间的差大于threshold,即要累加的数超过20个时候,将大任务分解成小任务
            int middle = (start+ end)/2;
            SumTask left = new SumTask(arry, start, middle);
            SumTask right = new SumTask(arry, middle, end);
            //并行执行两个 小任务
            left.fork();
            right.fork();
            //把两个小任务累加的结果合并起来
            return left.join()+right.join();
        }
        
    }
    
}
复制代码

 

执行结果:

初始化数组总和:1008
多线程执行结果:1008

 

分析

在Java 7中引入了一种新的线程池:ForkJoinPool。

它同ThreadPoolExecutor一样,也实现了Executor和ExecutorService接口。它使用了一个无限队列来保存需要执行的任务而线程的数量则是通过构造函数传入,如果没有向构造函数中传入希望的线程数量,那么当前计算机可用的CPU数量会被设置为线程数量作为默认值。

ForkJoinPool主要用来使用分治法(Divide-and-Conquer Algorithm)来解决问题。典型的应用比如快速排序算法。

这里的要点在于,ForkJoinPool需要使用相对少的线程来处理大量的任务。

比如要对1000万个数据进行排序,那么会将这个任务分割成两个500万的排序任务和一个针对这两组500万数据的合并任务。以此类推,对于500万的数据也会做出同样的分割处理,到最后会设置一个阈值来规定当数据规模到多少时,停止这样的分割处理。比如,当元素的数量小于10时,会停止分割,转而使用插入排序对它们进行排序。

那么到最后,所有的任务加起来会有大概2000000+个。问题的关键在于,对于一个任务而言,只有当它所有的子任务完成之后,它才能够被执行。

所以当使用ThreadPoolExecutor时,使用分治法会存在问题,因为ThreadPoolExecutor中的线程无法像任务队列中再添加一个任务并且在等待该任务完成之后再继续执行。而使用ForkJoinPool时,就能够让其中的线程创建新的任务,并挂起当前的任务,此时线程就能够从队列中选择子任务执行。

 

以上程序的关键是fork()和join()方法。在ForkJoinPool使用的线程中,会使用一个内部队列来对需要执行的任务以及子任务进行操作来保证它们的执行顺序。

那么使用ThreadPoolExecutor或者ForkJoinPool,会有什么性能的差异呢?

首先,使用ForkJoinPool能够使用数量有限的线程来完成非常多的具有父子关系的任务,比如使用4个线程来完成超过200万个任务。但是,使用ThreadPoolExecutor时,是不可能完成的,因为ThreadPoolExecutor中的Thread无法选择优先执行子任务,需要完成200万个具有父子关系的任务时,也需要200万个线程,显然这是不可行的。

ps:ForkJoinPool在执行过程中,会创建大量的子任务,导致GC进行垃圾回收,这些是需要注意的。

本文转载自:https://www.cnblogs.com/lixuwu/p/7979480.html

枯藤KT

枯藤KT

粉丝 4
博文 105
码字总数 62638
作品 0
昌平
程序员
私信 提问
Java并发编程(五) ForkJoinPool的使用

一.前言 之前在整理线程使用的时候,无意间看到了ForkJoinPool,在JDK1.7时(新)加入的,就学习了如何使用; 二. ForkJoinPool 使用 2.1 ForkJoinPool的使用姿势 ForkJoinPool采用工作窃取算法,将...

JackieRiver
2018/09/15
2.4K
0
Java线程之fork/join框架

fork/join框架是用多线程的方式实现分治法来解决问题。fork指的是将问题不断地缩小规模,join是指根据子问题的计算结果,得出更高层次的结果。 fork/join框架的使用有一定的约束条件: 1. 除...

legend3
2014/09/25
12
0
【转】JDK 7 中的 Fork/Join 模式

介绍 随着多核芯片逐渐成为主流,大多数软件开发人员不可避免地需要了解并行编程的知识。而同时,主流程序语言正在将越来越多的并行特性合并到标准库或者语言本身之中。我们可以看到,JDK 在...

mj4738
2011/11/28
357
0
浅谈JDK 1.8中的PrarllelStream

Stream是JDK 1.8中新增加的一个特性,就如同一个高级版迭代器(Iterator),可无限数据源,单向,不可往复,遍历过一次后即用尽了,正如流水一去不复返。而和迭代器只能命令式地、串行化操作不...

沐展
10/17
0
0
Java7中的ForkJoin并发框架初探(中)——JDK中实现简要分析

根据前文描述的Doug Lea的理论基础,在JDK1.7中已经给出了Fork Join的实现。在Java SE 7的API中,多了ForkJoinTask、ForkJoinPool、ForkJoinWorkerThread、RecursiveAction、RecursiveTask这...

Nori
2016/05/20
188
0

没有更多内容

加载失败,请刷新页面

加载更多

技术分享 | MySQL 8.0:字符集从 utf8 转换成 utf8mb4

作者:胡呈清 整理 MySQL 8.0 文档时发现一个变更:默认字符集由 latin1 变为 utf8mb4。想起以前整理过字符集转换文档,升级到 MySQL 8.0 后大概率会有字符集转换的需求,在此正好分享一下。...

爱可生
18分钟前
4
0
不管单机还是集群的限流实现已经给你准备好了

限流算法 计数器算法 维护一个counter,规定在单位时间内counter的大小不能超过最大值,每隔固定时间就将counter的值置零。如果这个counter大于设定的阈值,那么系统就拒绝请求 漏桶算法 维护...

阿提说说
29分钟前
4
0
文件管理

通过CLI登录进行文件管理 .表示当前目录,..表示父目录,具有隐藏文件。支持缩写与TAB键补全 1、目录操作 pwd#打印工作目录 cd <directory>#改变工作目录 dir [/all][<directory>]#查看目录内...

悠悠子佩
31分钟前
4
0
Netty学习笔记(10)——Netty中的Channel组件

1. Channel的功能 1. 与NIO中的Channel一样,它实现了网络操作的抽象类,聚合了一系列的网络IO功能,包括读写数据、建立连接、关闭连接等功能。通过外观模式,将数据读写、连接建立与断开等操...

江左煤郎
35分钟前
3
0
二叉树的深度

从根结点到叶结点依次经过的结点(含根、叶结点)形成树的一条路径,最长路径的长度为树的深度。 public int TreeDepth(TreeNode root) { return root == null ? 0 : 1 + Math.max(Tree...

Garphy
43分钟前
4
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部