快捷導(dǎo)航

Java并發(fā)編程之Fork/Join框架詳解

更新時(shí)間：2023年12月27日 09:14:05 作者：緣來(lái)如此09

這篇文章主要介紹了Java并發(fā)編程之Fork/Join框架詳解,Fork/Join框架是Java7提供的一個(gè)用于并行執(zhí)行任務(wù)的框架,是一個(gè)把大任務(wù)分割成若干個(gè)小任務(wù),最終匯總每個(gè)小任務(wù)結(jié)果后得到大任務(wù)結(jié)果的框架,需要的朋友可以參考下

一、簡(jiǎn)介

Fork/Join框架是Java7提供的一個(gè)用于并行執(zhí)行任務(wù)的框架，是一個(gè)把大任務(wù)分割成若干個(gè)小任務(wù)，最終匯總每個(gè)小任務(wù)結(jié)果后得到大任務(wù)結(jié)果的框架。

我們?cè)偻ㄟ^(guò)Fork和Join這兩個(gè)單詞來(lái)理解一下Fork/Join框架。

Fork就是把一個(gè)大任務(wù)切分為若干子任務(wù)并行的執(zhí)行，Join就是合并這些子任務(wù)的執(zhí)行結(jié)果，最后得到這個(gè)大任務(wù)的結(jié)果。

比如計(jì)算1+2+…+10000，可以分割成10個(gè)子任務(wù)，每個(gè)子任務(wù)分別對(duì)1000個(gè)數(shù)進(jìn)行求和，最終匯總這10個(gè)子任務(wù)的結(jié)果。

二、流程

1.分割任務(wù)：

首先我們需要有一個(gè)fork類來(lái)把大任務(wù)分割成子任務(wù)，有可能子任務(wù)還是很大，所以還需要不停地分割，直到分割出的子任務(wù)足夠小。

2.執(zhí)行任務(wù)并合并結(jié)果。

分割的子任務(wù)分別放在雙端隊(duì)列里，然后幾個(gè)啟動(dòng)線程分別從雙端隊(duì)列里獲取任務(wù)執(zhí)行。子任務(wù)執(zhí)行完的結(jié)果都統(tǒng)一放在一個(gè)隊(duì)列里，啟動(dòng)一個(gè)線程從隊(duì)列里拿數(shù)據(jù)，然后合并這些數(shù)據(jù)。

三、工作竊取算法

工作竊?。╳ork-stealing）算法是指某個(gè)線程從其他隊(duì)列里竊取任務(wù)來(lái)執(zhí)行。

假如我們需要做一個(gè)比較大的任務(wù)，可以把這個(gè)任務(wù)分割為若干互不依賴的子任務(wù)，為了減少線程間的競(jìng)爭(zhēng)，把這些子任務(wù)分別放到不同的隊(duì)列里，并為每個(gè)隊(duì)列創(chuàng)建一個(gè)單獨(dú)的線程來(lái)執(zhí)行隊(duì)列里的任務(wù)，線程和隊(duì)列一一對(duì)應(yīng)。

比如A線程負(fù)責(zé)處理A隊(duì)列里的任務(wù)。但是，有的線程會(huì)先把自己隊(duì)列里的任務(wù)干完，而其他線程對(duì)應(yīng)的隊(duì)列里還有任務(wù)等待處理。

干完活的線程與其等著，不如去幫其他線程干活，于是它就去其他線程的隊(duì)列里竊取一個(gè)任務(wù)來(lái)執(zhí)行。

而在這時(shí)它們會(huì)訪問(wèn)同一個(gè)隊(duì)列，所以為了減少竊取任務(wù)線程和被竊取任務(wù)線程之間的競(jìng)爭(zhēng)，通常會(huì)使用雙端隊(duì)列，被竊取任務(wù)線程永遠(yuǎn)從雙端隊(duì)列的頭部拿任務(wù)執(zhí)行，而竊取任務(wù)的線程永遠(yuǎn)從雙端隊(duì)列的尾部拿任務(wù)執(zhí)行

優(yōu)點(diǎn)：充分利用線程進(jìn)行并行計(jì)算，減少了線程間的競(jìng)爭(zhēng)。

缺點(diǎn)：在某些情況下還是存在競(jìng)爭(zhēng)，比如雙端隊(duì)列里只有一個(gè)任務(wù)時(shí)。并且該算法會(huì)消耗了更多的系統(tǒng)資源，比如創(chuàng)建多個(gè)線程和多個(gè)雙端隊(duì)列。

四、ForkJoin應(yīng)用

Fork/Join使用兩個(gè)類來(lái)完成分治的操作

1.ForkJoinTask

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

·RecursiveAction：用于沒(méi)有返回結(jié)果的任務(wù)。·RecursiveTask：用于有返回結(jié)果的任務(wù)核心方法：

fork()：在當(dāng)前線程運(yùn)行的線程池中創(chuàng)建一個(gè)子任務(wù)；join()：模塊子任務(wù)完成的時(shí)候返回任務(wù)結(jié)果；invoke()：執(zhí)行任務(wù)，也可以實(shí)時(shí)等待最終執(zhí)行結(jié)果

2.ForkJoinPool

線程池最大的特點(diǎn)就是分叉(fork)合并(join)模式，將一個(gè)大任務(wù)拆分成多個(gè)小任務(wù)，并行執(zhí)行，再結(jié)合工作竊取算法提高整體的執(zhí)行效率，充分利用CPU資源。

3.實(shí)例

 
    private static final Integer MAX = 200;
    static class MyForkJoinTask extends RecursiveTask<Integer> {
        // 子任務(wù)開(kāi)始計(jì)算的值
        private Integer startValue;
        // 子任務(wù)結(jié)束計(jì)算的值
        private Integer endValue;
        public MyForkJoinTask(Integer startValue , Integer endValue) {
            this.startValue = startValue;
            this.endValue = endValue;
        }
        @Override
        protected Integer compute() {
            // 如果條件成立，說(shuō)明這個(gè)任務(wù)所需要計(jì)算的數(shù)值分為足夠小了
            // 可以正式進(jìn)行累加計(jì)算了
            if(endValue - startValue < MAX) {
                System.out.println("開(kāi)始計(jì)算的部分：startValue = " + startValue + ";endValue = " + endValue);
                Integer totalValue = 0;
                for(int index = this.startValue ; index <= this.endValue  ; index++) {
                    totalValue += index;
                }
                return totalValue;
            }
            // 否則再進(jìn)行任務(wù)拆分，拆分成兩個(gè)任務(wù)
            else {
                MyForkJoinTask subTask1 = new MyForkJoinTask(startValue, (startValue + endValue) / 2);
                subTask1.fork();
                MyForkJoinTask subTask2 = new MyForkJoinTask((startValue + endValue) / 2 + 1 , endValue);
                subTask2.fork();
                return subTask1.join() + subTask2.join();
            }
        }
    }
    public static void main(String[] args) {
        // 這是Fork/Join框架的線程池
        ForkJoinPool pool = new ForkJoinPool();
        ForkJoinTask<Integer> taskFuture =  pool.submit(new MyForkJoinTask(1,1001));
        try {
            Integer result = taskFuture.get();
            System.out.println("result = " + result);
        } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace(System.out);
        }
    }

到此這篇關(guān)于Java并發(fā)編程之Fork/Join框架詳解的文章就介紹到這了,更多相關(guān)Fork/Join框架內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: