Java 并发编程之ForkJoin框架

目录
  • 1、什么是ForkJoin框架
  • 2、ForkJoinTask
  • 3、ForkJoinPool
  • 4、打印斐波那契数列
  • 5、ForkJoin归并排序
  • 总结

1、什么是ForkJoin框架

ForkJoin框架是java的JUC包里提供的,用于处理一些比较繁重的任务,会将这个大任务分为多个小任务,多个小任务处理完成后会将结果汇总给Result,体现的是一种“分而治之”的思想。第一步,拆分fork任务,将大任务分为多个小任务;第二步,归并join,会将小任务的处理结果进行归并为一个结果。

2、ForkJoinTask

ForkJoinTask是ForkJoin框架的提供的任务API,ForkJoinTask是一个抽象类,有两个主要的实现类,RecursiveTask和RecursiveAction,其中RecursiveTask和RecursiveAction的主要区别是,RecursiveAction没有返回值,而RecursiveTask是有返回值的

3、ForkJoinPool

ForkJoinPool类是forkjoin框架的线程池实现,基于ExecutorService接口。这个线程池是jdk1.7才加入的,用于管理线程,执行forkjoin的任务。对于线程池的使用,我们使用ThreadPoolExecutor比较多,可以在idea里看一下uml类图,可以看出ForkJoinPool和ThreadPoolExecutor实现差不多的。

ForkJoinPool()
ForkJoinPool(int parallelism)
ForkJoinPool(int parallelism, ForkJoinWorkerThreadFactory factory,
 UncaughtExceptionHandler handler, boolean asyncMode)

几个重要的参数:

  • parallelism:并行度,并行执行线程,可用指定,也可以不指定,不指定的情况,是根据cpu核数创建可用的线程
  • ForkJoinWorkerThreadFactory:创建线程的工厂实现
  • UncaughtExceptionHandler :因为未知异常中断的回调处理
  • asyncMode:是否异步,默认情况是false

使用时候,可以直接创建ForkJoinPool,可以不传参,不传参的情况,默认指定的线程并行数为Runtime.getRunTime().availableProcessors();,表示根据cpu核数创建可用线程数

ForkJoinPool forkJoinPool = new ForkJoinPool();
ArraySortTask task = new ArraySortTask(array , 0 , size);
forkJoinPool.submit(task);
task.get();

也是可用传参,对并行度进行指定的public ForkJoinPool(int parallelism), parallelism并行度,并行执行几个线程

将forkjoin任务加入到FrokJoinPool线程池有几种方式

  • execute():调用其 fork 方法在多个线程之间拆分工作。
  • invoke():在ForkJoinPool线程池上调用invoke方法
  • submit():返回一个Future对象,Future可以进行监控,任务完成后返回结果

4、打印斐波那契数列

ForkJoin框架可以用于一些递归遍历的场景,对于斐波那契数列,你可以比较熟悉,因为在面试中有时候经常问到,斐波那契数列的特点就是最后一项的结果等于前面两项的和

package com.example.concurrent.forkjoin;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.ForkJoinTask;
import java.util.concurrent.RecursiveTask;
/**
 * <pre>
 *      斐波那契数列
 * </pre>
 * <p>
 * <pre>
 * @author nicky.ma
 * 修改记录
 *    修改后版本:     修改人:  修改日期: 2021/10/12 16:22  修改内容:
 * </pre>
 */
public class Fibonacci extends RecursiveTask<Integer>{
    private int n;
    public Fibonacci(int n) {
        this.n = n;
    }
    @Override
    protected Integer compute() {
        if (n <= 1)
            return n;
        Fibonacci f1 = new Fibonacci(n - 1);
        f1.fork();
        Fibonacci f2 = new Fibonacci(n - 2);
        f2.fork();
        return f1.join() + f2.join();
    }
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        ForkJoinPool pool = new ForkJoinPool();
        for (int i = 0; i< 10; i++) {
            ForkJoinTask task = pool.submit(new Fibonacci(i));
            System.out.println(task.get());
        }
    }
}

5、ForkJoin归并排序

面试题:快速实现对一个长度百万的数组的排序

难点:可以使用归并排序,多线程如何组织实现归并排序

package com.example.concurrent.forkjoin;
import java.util.Arrays;
import java.util.Random;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.RecursiveAction;
/**
 * <pre>
 *      大数组排序
 * </pre>
 * <p>
 * <pre>
 * @author mazq
 * 修改记录
 *    修改后版本:     修改人:  修改日期: 2021/10/12 17:04  修改内容:
 * </pre>
 */
public class ArraySortTask extends RecursiveAction{
    final long[] array; final int lo, hi;
    ArraySortTask(long[] array, int lo, int hi) {
        this.array = array; this.lo = lo; this.hi = hi;
    }
    ArraySortTask(long[] array) { this(array, 0, array.length); }
    @Override
    protected void compute() {
        if (hi - lo < THRESHOLD)
            // 少于阀值,使用Arrays.sort 快排
            sortSequentially(lo, hi);
        else {
            /* 归并排序 */
            // 取中间值
            int mid = (lo + hi) >>> 1;
            // 拆分任务
            invokeAll(new ArraySortTask(array, lo, mid),
                    new ArraySortTask(array, mid, hi));
            // 归并结果
            merge(lo, mid, hi);
        }
    }
    // implementation details follow:
    static final int THRESHOLD = 1000;
    void sortSequentially(int lo, int hi) {
        Arrays.sort(array, lo, hi);
    }
    void merge(int lo, int mid, int hi) {
        long[] buf = Arrays.copyOfRange(array, lo, mid);
        for (int i = 0, j = lo, k = mid; i < buf.length; j++)
            array[j] = (k == hi || buf[i] < array[k]) ?
                    buf[i++] : array[k++];
    }
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        int size = 10_000;
        long[] array = new long[size];
        Random random = new Random();
        for (int i = 0; i< size; i++) {
            array[i] = random.nextInt();
        }
        ForkJoinPool forkJoinPool = new ForkJoinPool();
        ArraySortTask task = new ArraySortTask(array , 0 , size);
        forkJoinPool.submit(task);
        task.get();
        for (long a : array) {
            System.out.println(a);
        }
    }
}

总结

本篇文章就到这里了,希望能够给你带来帮助,也希望您能够多多关注我们的更多内容!

(0)

相关推荐

  • Java ForkJoin框架的原理及用法

    这篇文章主要介绍了Java ForkJoin框架的原理及用法,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下 ForkJoin分析 一.ForkJoin ForkJoin是由JDK1.7后提供多线并发处理框架.ForkJoin的框架的基本思想是分而治之.什么是分而治之?分而治之就是将一个复杂的计算,按照设定的阈值进行分解成多个计算,然后将各个计算结果进行汇总.相应的ForkJoin将复杂的计算当做一个任务.而分解的多个计算则是当做一个子任务. 二

  • java中的forkjoin框架的使用

    fork join框架是java 7中引入框架,这个框架的引入主要是为了提升并行计算的能力. fork join主要有两个步骤,第一就是fork,将一个大任务分成很多个小任务,第二就是join,将第一个任务的结果join起来,生成最后的结果.如果第一步中并没有任何返回值,join将会等到所有的小任务都结束. 还记得之前的文章我们讲到了thread pool的基本结构吗? ExecutorService - ForkJoinPool 用来调用任务执行. workerThread - ForkJoi

  • java8中forkjoin和optional框架使用

    并行流与串行流 并行流就是把一个内容分成多个数据块,并用不同的线程分别处理每个数据块的流. java 8 中将并行进行了优化,我们可以很容易的对数据进行并行操作.Stream API 可以声明性地通过 parallel()与 sequential()在并行流与顺序流之间进行切换. 了解 Fork/Join 框架 Fork/Join 框架:就是在必要的情况下,将一个大任务,进形拆分(fork)成若干个小任务(拆到不可再拆时),再将一个个的小任务运行的结果进行join汇总. Fork/Join 框架

  • Java 并发编程之ForkJoin框架

    目录 1.什么是ForkJoin框架 2.ForkJoinTask 3.ForkJoinPool 4.打印斐波那契数列 5.ForkJoin归并排序 总结 1.什么是ForkJoin框架 ForkJoin框架是java的JUC包里提供的,用于处理一些比较繁重的任务,会将这个大任务分为多个小任务,多个小任务处理完成后会将结果汇总给Result,体现的是一种"分而治之"的思想.第一步,拆分fork任务,将大任务分为多个小任务:第二步,归并join,会将小任务的处理结果进行归并为一个结果.

  • Java并发编程之Fork/Join框架的理解

    一.Fork/Join框架的理解 ForkJoinTask类属于java.util.concurrent 包下: ForkJoinTask类下有2个子类,分别为RecursiveTask和RecursiveAction类:(lz示例中使用RecursiveTask类进行重写compute()方法进行实现数值的累加计算) ForkJoinTask类 将一个大的任务拆分成多个子任务进行并行处理,最后将子任务结果合并成最后的计算结果,并进行输出. 二.Fork/Join框架使用示例 示例场景:对数值进

  • Java并发编程之JUC并发核心AQS同步队列原理剖析

    目录 一.AQS介绍 二.AQS中的队列 1.同步等待队列 2.条件等待队列 3.AQS队列节点Node 三.同步队列源码分析 1.同步队列分析 2.同步队列--独占模式源码分析 3.同步队列--共享模式源码分析 一.AQS介绍 队列同步器AbstractQueuedSynchronizer(简称AQS),AQS定义了一套多线程访问共享资源的同步器框架,是用来构建锁或者其他同步组件的基础框架,是一个依赖状态(state)的同步器.Java并发编程的核心在java.util.concurrent(

  • Java并发编程之Semaphore(信号量)详解及实例

    Java并发编程之Semaphore(信号量)详解及实例 概述 通常情况下,可能有多个线程同时访问数目很少的资源,如客户端建立了若干个线程同时访问同一数据库,这势必会造成服务端资源被耗尽的地步,那么怎样能够有效的来控制不可预知的接入量呢?及在同一时刻只能获得指定数目的数据库连接,在JDK1.5 java.util.concurrent 包中引入了Semaphore(信号量),信号量是在简单上锁的基础上实现的,相当于能令线程安全执行,并初始化为可用资源个数的计数器,通常用于限制可以访问某些资源(物

  • java并发编程之cas详解

    CAS(Compare and swap)比较和替换是设计并发算法时用到的一种技术.简单来说,比较和替换是使用一个期望值和一个变量的当前值进行比较,如果当前变量的值与我们期望的值相等,就使用一个新值替换当前变量的值.这听起来可能有一点复杂但是实际上你理解之后发现很简单,接下来,让我们跟深入的了解一下这项技术. CAS的使用场景 在程序和算法中一个经常出现的模式就是"check and act"模式.先检查后操作模式发生在代码中首先检查一个变量的值,然后再基于这个值做一些操作.下面是一个

  • Java并发编程之Condition源码分析(推荐)

    Condition介绍 上篇文章讲了ReentrantLock的加锁和释放锁的使用,这篇文章是对ReentrantLock的补充.ReentrantLock#newCondition()可以创建Condition,在ReentrantLock加锁过程中可以利用Condition阻塞当前线程并临时释放锁,待另外线程获取到锁并在逻辑后通知阻塞线程"激活".Condition常用在基于异步通信的同步机制实现中,比如dubbo中的请求和获取应答结果的实现. 常用方法 Condition中主要的

  • 浅谈Java并发编程之Lock锁和条件变量

    简单使用Lock锁 Java 5中引入了新的锁机制--java.util.concurrent.locks中的显式的互斥锁:Lock接口,它提供了比synchronized更加广泛的锁定操作.Lock接口有3个实现它的类:ReentrantLock.ReetrantReadWriteLock.ReadLock和ReetrantReadWriteLock.WriteLock,即重入锁.读锁和写锁.lock必须被显式地创建.锁定和释放,为了可以使用更多的功能,一般用ReentrantLock为其实例

  • 深入分析Java并发编程之CAS

    在Java并发编程的世界里,synchronized 和 Lock 是控制多线程并发环境下对共享资源同步访问的两大手段.其中 Lock 是 JDK 层面的锁机制,是轻量级锁,底层使用大量的自旋+CAS操作实现的. 学习并发推荐<Java并发编程的艺术> 那什么是CAS呢?CAS,compare and swap,即比较并交换,什么是比较并交换呢?在Lock锁的理念中,采用的是一种乐观锁的形式,即多线程去修改共享资源时,不是在修改之前就加锁,而是乐观的认为没有别的线程和自己争锁,就是通过CAS的

  • Java并发编程之ReadWriteLock读写锁的操作方法

    1.ReadWriteLock介绍 为什么我们有了Lock,还要用ReadWriteLock呢.我们对共享资源加锁之后,所有的线程都将会等待.Lock读操作也锁,写操作也会锁,而对共享资源读的时候,其实是不用加锁的.当然读写同时存在的情况也会有. 比如我们数据库常用操作有增删改查,增删改都是写操作,写操作必须加锁,而读操作可以共享.不是所有的操作都需要加锁. 为了进一步提高复用性和粒度,写操作独占,读操作共享,不加锁. ReadWriteLock管理一组锁,一个是只读的锁,一个是写锁.读锁可以在

  • Java并发编程之ReentrantLock可重入锁的实例代码

    目录 1.ReentrantLock可重入锁概述2.可重入3.可打断4.锁超时5.公平锁6.条件变量 Condition 1.ReentrantLock可重入锁概述 相对于 synchronized 它具备如下特点 可中断 synchronized锁加上去不能中断,a线程应用锁,b线程不能取消掉它 可以设置超时时间 synchronized它去获取锁时,如果对方持有锁,那么它就会进入entryList一直等待下去.而可重入锁可以设置超时时间,规定时间内如果获取不到锁,就放弃锁 可以设置为公平锁

随机推荐