多线程 ForkJoinPool

背景: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进行垃圾回收,这些是需要注意的。

未完待续。。。

时间: 2024-11-02 05:56:42

多线程 ForkJoinPool的相关文章

java多线程 -- ForkJoinPool 分支/ 合并框架 工作窃取

Fork/Join 框架:就是在必要的情况下,将一个大任务,进行拆分(fork)成若干个小任务(拆到不可再拆时),再将一个个的小任务运算的结果进行 join 汇总. Fork/Join 框架与线程池的区别 采用 "工作窃取"模式(work-stealing):当执行新的任务时它可以将其拆分分成更小的任务执行,并将小任务加到线程队列中,然后再从一个随机线程的队列中偷一个并把它放在自己的队列中. 相对于一般的线程池实现,fork/join框架的优势体现在对其中包含的任务的处理方式上.在一般

多线程-Thread-Runnable

一.多线程     1.基本概念         进程:正在运行中的程序,一个进程中至少包含一个线程         线程:进程的任务,执行任务的一个通道,一个进程中可以包含多个线程     2.多线程执行的特点:         两种方式:分时调度/抢占式调度(java属于抢占) 二.Thread 类(java.lang)     1.概述:使用该类表示多线程对象,只要创建一个Thread对象,就是一个线程对象产生了     2.定义:public class Thread extends O

【转】 Java 多线程之一

转自   Java 多线程 并发编程 一.多线程 1.操作系统有两个容易混淆的概念,进程和线程. 进程:一个计算机程序的运行实例,包含了需要执行的指令:有自己的独立地址空间,包含程序内容和数据:不同进程的地址空间是互相隔离的:进程拥有各种资源和状态信息,包括打开的文件.子进程和信号处理. 线程:表示程序的执行流程,是CPU调度执行的基本单位:线程有自己的程序计数器.寄存器.堆栈和帧.同一进程中的线程共用相同的地址空间,同时共享进进程锁拥有的内存和其他资源. 2.Java标准库提供了进程和线程相关

Java 多线程 并发编程 (转)

一.多线程 1.操作系统有两个容易混淆的概念,进程和线程. 进程:一个计算机程序的运行实例,包含了需要执行的指令:有自己的独立地址空间,包含程序内容和数据:不同进程的地址空间是互相隔离的:进程拥有各种资源和状态信息,包括打开的文件.子进程和信号处理. 线程:表示程序的执行流程,是CPU调度执行的基本单位:线程有自己的程序计数器.寄存器.堆栈和帧.同一进程中的线程共用相同的地址空间,同时共享进进程锁拥有的内存和其他资源. 2.Java标准库提供了进程和线程相关的API,进程主要包括表示进程的jav

Java 从单核到多核的多线程(并发)

JAVA 并发       java的并行编程比较复杂,我也理解不深.但是最近由于要并行训练分类器,琢磨了一点,有错误请指正.只是大体介绍一下而已. 很多问题我们使用顺序编程便可以解决,但是有些问题如果能够使用多线程并行的执行其中的任务则可以很大程度的提高时间效率,所以多线程还是很有必要的. 我自己总结了JAVA并行的3个发展阶段(菜鸟总结,请体谅) 第一阶段:Thread ,Runable 第二阶段:Executor框架 第三阶段: ForkJoin并行框架 并发很大一方面是为了提高程序运行速

Java 多线程 并发编程

转自: http://blog.csdn.net/escaflone/article/details/10418651 一.多线程 1.操作系统有两个容易混淆的概念,进程和线程. 进程:一个计算机程序的运行实例,包含了需要执行的指令:有自己的独立地址空间,包含程序内容和数据:不同进程的地址空间是互相隔离的:进程拥有各种资源和状态信息,包括打开的文件.子进程和信号处理. 线程:表示程序的执行流程,是CPU调度执行的基本单位:线程有自己的程序计数器.寄存器.堆栈和帧.同一进程中的线程共用相同的地址空

java8新特性(六):Stream多线程并行数据处理

转:http://blog.csdn.net/sunjin9418/article/details/53143588 将一个顺序执行的流转变成一个并发的流只要调用 parallel()方法 public static long parallelSum(long n){    return Stream.iterate(1L, i -> i +1).limit(n).parallel().reduce(0L,Long::sum); } 并行流就是一个把内容分成多个数据块,并用不不同的线程分别处理每

java 多线程--- Thread Runnable Executors

java 实现多线程的整理: Thread实现多线程的两种方式: (1)继承 Thread类,同时重载 run 方法: class PrimeThread extends Thread { long minPrime; primeThread(long minPrime) { this.minPrime = minPrime; } public void run() { // compute primes larger than minPrime } } PrimeThread p = new

Java读书笔记(4)-多线程(二)

2016-1-2 线程通信 传统的线程通信 Object类提供了wait(),notify()和notifyAll三个方法 适用情况:synchronized修饰的同步方法或者synchronized方法 wait():导致当前线程等待,直到其他线程调用该同步监视器的notify()或notifyAll方法来唤醒该线程,调用wait方法后本线程会释放对同步监视器的锁定 notify():唤醒在此同步监视器上等待的单个线程.如果有多个线程在等待,则随机唤醒其中一个 notifyAll():唤醒在此