快捷導(dǎo)航

Java多線程高并發(fā)中的Fork/Join框架機(jī)制詳解

更新時(shí)間：2021年11月16日 10:34:02 作者：張起靈-小哥

本文主要介紹了 Java 多線程高并發(fā)中的 Fork/Join 框架的基本原理和其使用的工作竊取算法(work-stealing)、設(shè)計(jì)方式和部分實(shí)現(xiàn)源碼，感興趣的朋友跟隨小編一起看看吧

1.Fork/Join框架簡(jiǎn)介

Fork/Join 它可以將一個(gè)大的任務(wù)拆分成多個(gè)子任務(wù)進(jìn)行并行處理，最后將子任務(wù)結(jié)果合并成最后的計(jì)算結(jié)果，并進(jìn)行輸出。Fork/Join 框架要完成兩件事情：

Fork：把一個(gè)復(fù)雜任務(wù)進(jìn)行分拆，大事化小：把一個(gè)復(fù)雜任務(wù)進(jìn)行分拆，大事化小

Join：把分拆任務(wù)的結(jié)果進(jìn)行合并

在 Java 的 Fork/Join 框架中，使用兩個(gè)類完成上述操作：

ForkJoinTask: 我們要使用 Fork/Join 框架，首先需要?jiǎng)?chuàng)建一個(gè) ForkJoin 任務(wù)。該類提供了在任務(wù)中執(zhí)行 fork 和 join 的機(jī)制。通常情況下我們不需要直接集成 ForkJoinTask 類，只需要繼承它的子類，F(xiàn)ork/Join 框架提供了兩個(gè)子類：

RecursiveAction：用于沒有返回結(jié)果的任務(wù)
RecursiveTask:用于有返回結(jié)果的任務(wù)

ForkJoinPool: ForkJoinTask 需要通過 ForkJoinPool 來執(zhí)行。

RecursiveTask: 繼承后可以實(shí)現(xiàn)遞歸(自己調(diào)自己)調(diào)用的任務(wù)。

可以在jdk官方文檔中看到：

2.簡(jiǎn)單應(yīng)用

實(shí)現(xiàn)從 1 + 2 + ... + 100 ，將它們拆分成多個(gè)小任務(wù)，分別求和，最終再將這些結(jié)果合并。

這里就是參照官方文檔，先繼承RecursiveTask類，重寫其中的compute方法，然后定義有參構(gòu)造，而ForkJoinTask需要通過 ForkJoinPool 來執(zhí)行，所以還需要?jiǎng)?chuàng)建分支合并池ForkJoinPool對(duì)象。

package test.forkjoin;
 
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.ForkJoinTask;
import java.util.concurrent.RecursiveTask;
 
/**
 *
 */
class MyTask extends RecursiveTask<Integer> {
    //拆分差值不能超過10，每次計(jì)算10以內(nèi)的連加操作
    private static final Integer NUMBER = 10;
    private int begin; //拆分左區(qū)間的值
    private int end; //拆分右區(qū)間的值
    private int result; //最終結(jié)果
 
    public MyTask(int begin,int end) {
        this.begin = begin;
        this.end = end;
    }
 
    @Override
    protected Integer compute() {
        //判斷區(qū)間差值是否大于10
        if ((end - begin) <= NUMBER) {
            //將區(qū)間內(nèi)的值依次相加
            for (int i = begin; i <= end; i++) {
                result += i;
            }
        } else { //區(qū)間差值大于10，進(jìn)一步拆分
            //獲取中間值
            int middle = (begin + end) / 2;
            //拆分的左區(qū)間
            MyTask taskLeft = new MyTask(begin,middle);
            //拆分的右區(qū)間
            MyTask taskRight = new MyTask(middle + 1,end);
            //fork方法進(jìn)行拆分
            taskLeft.fork();
            taskRight.fork();
            //join方法進(jìn)行合并
            result = taskLeft.join() + taskRight.join();
        }
        return result;
    }
}
 
public class ForkJoinDemo {
    public static void main(String[] args) {
        //創(chuàng)建MyTask對(duì)象
        MyTask myTask = new MyTask(1,100);
        //創(chuàng)建分支合并池對(duì)象
        ForkJoinPool forkJoinPool = new ForkJoinPool();
        ForkJoinTask<Integer> forkJoinTask = forkJoinPool.submit(myTask);
        try {
            //獲取最終合并后的結(jié)果
            Integer ans = forkJoinTask.get();
            System.out.println(ans);
        } catch (Exception e) {
            e.printStackTrace();
        }finally {
            //關(guān)閉池對(duì)象
            forkJoinPool.shutdown();
        }
    }
}