Fork-Join框架

简介: 在JDK1.7引入了一种新的并行编程模式“fork-join”,它是实现了“分而治之”思想的Java并发编程框架。网上关于此框架的各种介绍很多,本文从框架特点出发,通过几个例子来进行实用性的介绍。

1 fork-join框架的特点

它对问题的解决思路是分而治之,先将一个问题fork(分为)几个子问题,然后子问题又分为孙子问题,直至细分为一个容易计算的问题,然后再将结果依次join(结合)为最终的答案.是不是感觉和云计算中的Map-reduce计算模型很像?思路是一样的,只不过fork-join运行在一个JVM中的多个线程内,而map-reduce运行在分布式计算节点上


在运行线程时,它使用“work-steal”(任务偷取)算法.一般来说,fork-join会启动多个线程(由参数指定,若不指定则默认为CPU核心数量),每个线程负责一个任务队列,并依次从队列头部获得任务并执行.当某个线程空闲时,它会从其他线程的任务队列尾部偷取一个任务来执行,这样就保证了线程的运行效率达到最高。



它面向的问题域是可以大量并行执行的计算任务,例如计算某个大型数组中每个元素的平方(当然这个有些无趣),其计算对象最好是一些独立的元素,不会被其他线程访问,也没有同步、互斥要求,更不要涉及IO或者无限循环.当然此框架也可以执行普通的并发编程任务,但是这时就失去了性能优势


细分的计算任务有一个粗略的优化标准,即可以在100~10000条指令中执行完毕


了解以上思路后,来看看fork-join框架提供的几个工具类:

ForkJoinPool:支持fork-join框架的线程池,所有ForkJoinTask任务都必须在其中运行,线程池主要使用invoke()、invokeAll()等方法来执行任务,当然也可以使用原有的execute()和submit()方法;


ForkJoinTask:支持fork-join框架的任务抽象类,它是Future接口,它代表一个支持fork()和join()方法的任务;


RecursiveAction:ForkJoinTask的两个具体子类之一,代表没有返回值的ForkJoinTask任务;


RecursiveTask:ForkJoinTask的两个具体子类之一,代表有返回值的ForkJoinTask任务。


2 RecursiveAction

先来看一个使用RecursiveAction的例子,这段代码的目的是计算一个大型数组中每个元素x的一个公式的值,这个公式是sin(x)+cos(x)+tan(x):

public class RecursiveActionExam {
    private final static int NUMBER = 10000000;
    public static void main(String[] args) {
        double[] array = new double[NUMBER];
        for (int i = 0; i < NUMBER; i++) {
            array[i] = i;
        }
        long startTime = System.currentTimeMillis();
        System.out.println(Runtime.getRuntime().availableProcessors());
        ForkJoinPool forkJoinPool = new ForkJoinPool();
        forkJoinPool.invoke(new ComputeTask(array, 0, array.length));
        long endTime = System.currentTimeMillis();
        System.out.println("Time span = " + (endTime - startTime));
    }
}
class ComputeTask extends RecursiveAction {
    final double[] array;
    final int lo, hi;
    ComputeTask(double[] array, int lo, int hi) {
        this.array = array;
        this.lo = lo;
        this.hi = hi;
    }
    protected void compute() {
        if (hi - lo < 2) {
            for (int i = lo; i < hi; ++i)
                array[i] = Math.sin(array[i]) + Math.cos(array[i]) + Math.tan(array[i]);
        } else {
            int mid = (lo + hi) >>> 1;
            invokeAll(new ComputeTask(array, lo, mid),
                    new ComputeTask(array, mid, hi));
        }
    }
}

再看看单线程的情况:

public class RecursiveSequenceExam {
    private final static int NUMBER = 10000000;
    public static void main(String[] args) {
        double[] array = new double[NUMBER];
        for (int i = 0; i < NUMBER; i++) {
            array[i] = i;
        }
        long startTime = System.currentTimeMillis();
        for (int i = 0; i < NUMBER; i++) {
            array[i] = Math.sin(array[i]) + Math.cos(array[i]) + Math.tan(array[i]);
        }
        long endTime = System.currentTimeMillis();
        System.out.println("Time span = " + (endTime - startTime));
    }
}

运行结果是Time span = 12030。

由于我的CPU是4核的,再看看4线程的情况:

public class Recusive4ThreadExam {
    private final static int NUMBER = 10000000;
    public static void main(String[] args) throws InterruptedException {
        double[] array = new double[NUMBER];
        for (int i = 0; i < NUMBER; i++) {
            array[i] = i;
        }
        long startTime = System.currentTimeMillis();
        ExecutorService service = Executors.newFixedThreadPool(4);
        service.execute(new ArrayTask(array, 0, NUMBER / 4));
        service.execute(new ArrayTask(array, NUMBER / 4, NUMBER / 2));
        service.execute(new ArrayTask(array, NUMBER / 2, NUMBER*3 / 4));
        service.execute(new ArrayTask(array, NUMBER*3 / 4, NUMBER ));
        service.shutdown();
        service.awaitTermination(1,TimeUnit.DAYS);
        long endTime = System.currentTimeMillis();
        System.out.println("Time span = " + (endTime - startTime));
    }
}
class ArrayTask implements Runnable {
    final double[] array;
    final int lo, hi;
    ArrayTask(double[] array, int lo, int hi) {
        this.array = array;
        this.lo = lo;
        this.hi = hi;
    }
    @Override
    public void run() {
        for (int i = lo; i < hi; ++i)
            array[i] = Math.sin(array[i]) + Math.cos(array[i]) + Math.tan(array[i]);
    }
}

运行结果是Time span = 4064。可以看出由于fork-join框架采用了任务偷取算法,比普通4线程快了一点点。

3 RecursiveTask

下面来看一个更有意义的场景,寻找一个大型数组的最小值,这里我使用RecursiveTask(其实使用RecursiveAction也行,在它内部用一个成员变量保存结果即可)。代码如下:

public class RecursiveFindMax {
    private static Random rand = new Random(47);
    private static final int NUMBER = 1000000;
    public static void main(String[] args) {
        double[] array = new double[NUMBER];
        for (int i = 0; i < NUMBER; i++) {
            array[i] = rand.nextDouble();
        }
        ForkJoinPool pool = new ForkJoinPool();
        TaskFindMax task = new TaskFindMax(0, array.length - 1, array);
        pool.invoke(task);
        System.out.println("MaxValue = " + task.join());
    }
}
class TaskFindMax extends RecursiveTask<Double> {
    private final int lo;
    private final int hi;
    private final double[] array;
    //you can change THRESHOLD to get better efficiency
    private final static int THRESHOLD = 10;
    TaskFindMax(int lo, int hi, double[] array) {
        this.lo = lo;
        this.hi = hi;
        this.array = array;
    }
    @Override
    protected Double compute() {
        if ((hi - lo) < THRESHOLD) {
            double max = array[lo];
            for (int i = lo; i < hi; i++) {
                max = Math.max(max, array[i + 1]);
            }
            return max;
        } else {
            int mid = (lo + hi) >>> 1;
            TaskFindMax lhs = new TaskFindMax(lo, mid, array);
            TaskFindMax rhs = new TaskFindMax(mid, hi, array);
            invokeAll(lhs, rhs);
            return Math.max(lhs.join(), rhs.join());
        }
    }
}

pool.invoke(task)将一个最初的任务扔进了线程池执行,这个任务将会执行它的compute()方法。在此方法中,若满足某个条件(例如数组上界和下界只差小于阈值THRESHOLD)则直接在这一段数组中查找最大值;若不满足条件,则找出中值mid,然后new出两个子任务lhs(left hand side)和rhs(right hand side),并调用invokeAll(lhs, rhs)将这两个子任务都扔进线程池执行。任务的join()方法会得到返回值,若任务尚未执行完毕则会在此阻塞。

通过这种编程模式,很好的将递归思想用到了多线程领域。值得注意的是,通过调整THRESHOLD可以增加或减少任务的个数,从而极大的影响线程的执行。在很多情况下,使用fork-join框架并不会比普通的多线程效率更高,甚至比单线程运行效率更低。因此,必须找到适合的场景,然后进行多次调优,才能获得性能的改进。


小结

执行者与线程池的引入是因为Concurrency包的设计者想将线程的创建、执行和调度分离,从而使得用户能够更加专注于业务逻辑;Callable接口和Future接口使得异步执行结果的获取更加简单;ScheduledExecutorService取代Timer成为了线程重复和延迟执行的新标准;TimeUnit类的引入简化了时间段的表达工作;包中提供的五种线程池可以极大的满足程序员的各种需求,极端情况下还可以利用ThreadPoolExecutor类自己定制线程池。最后,从JDK1.7后引入的Fork-Join框架将“分而治之”的递归思想实现到线程池中,并应用“work-steal”算法实现了任务执行效率的提升






目录
相关文章
|
6月前
|
JavaScript Java 大数据
分享Fork/Join经典案例
`shigen`是位专注于Java、Python、Vue和Shell的博主,分享技术成长和认知。上篇文章探讨了Java的Fork/Join框架,它类似线程池,通过拆分大任务并并行执行提升效率。以大序列求和为例展示了ForkJoinPool的使用,与普通线程池对比,Fork/Join效率提升约50%。适合递归任务、独立子任务和长执行时间的任务。注意任务粒度、避免共享状态和死锁。推荐观看相关视频深入理解。一起学习,每天进步!
43 0
分享Fork/Join经典案例
|
6月前
|
并行计算 安全 算法
探索Java并发编程:Fork/Join框架的应用与优化
在多核处理器普及的今天,并发编程已经成为提高程序性能的重要手段。Java提供了多种并发工具,其中Fork/Join框架是处理分治任务的强大工具。本文将深入探讨Fork/Join框架的核心原理、使用场景以及性能优化技巧,帮助开发者更好地利用这一框架解决实际问题。通过实例分析,我们将看到如何有效地使用Fork/Join框架来加速计算密集型任务,并提供一系列最佳实践,以确保高效和线程安全的并发执行。
|
6月前
|
并行计算 算法 Java
Java并发 -- Fork/Join框架
Java并发 -- Fork/Join框架
61 0
|
并行计算 负载均衡 算法
什么是 Fork/Join 框架?Java 中如何使用 Fork/Join 框架?
什么是 Fork/Join 框架?Java 中如何使用 Fork/Join 框架?
|
算法
Fork/Join框架的学习和浅析
Fork/Join框架的学习和浅析
82 0
|
分布式计算 算法 Java
【JUC基础】16. Fork Join
“分而治之”一直是一个非常有效的处理大量数据的方法。著名的MapReduce也是采取了分而治之的思想。。简单地说,就是如果你要处理 1000 个数据,但是你并不具备处理 1000个数据的能力,那么你可以只处理其中的 10 个,然后分阶段处理 100 次,将 100 次的结进行合成,就是最终想要的对原始 1000 个数据的处理结果。而这就是Fork Join的基本思想。
108 1
|
并行计算 算法 Java
Fork/Join解读
Fork/Join解读
|
并行计算 算法 Java
【JAVA并发编程专题】Fork/Join框架的理解和使用
【JAVA并发编程专题】Fork/Join框架的理解和使用
|
缓存 并行计算 算法
浅析 Fork/Join 基本概念和实战
在 JDK 1.7 版本中提供了 Fork/Join 并行执行任务框架,它主要的作用是把大任务分割成若干个小任务,再对每个小任务得到的结果进行汇总,此种开发方法也叫做分治编程,分治编程可以极大的利用 CPU 资源,提高任务执行效率。
218 0
浅析 Fork/Join 基本概念和实战
|
并行计算 算法
Fork/Join框架简介
Fork/Join框架是Java7提供了的一个用于并行执行任务的框架, 是一个把大任务分割成若干个小任务,最终汇总每个小任务结果后得到大任务结果的框架。
260 0
Fork/Join框架简介