??什么是Forkjoin

Fork/Join 是一種在多線程領(lǐng)域中常用的算法或技術(shù)，它的核心思想是將大任務(wù)分割成若干個(gè)小任務(wù)，然后將這些小任務(wù)分配給多個(gè)線程并行處理，最終將結(jié)果合并起來。這種思想可以應(yīng)用于多種場景，例如圖像處理、批處理、并行排序等。

在 Java 中，F(xiàn)ork/Join 這種思想被封裝在了 java.util.concurrent 包中的 ForkJoinPool 類和 RecursiveTask 類中。ForkJoinPool 類是一個(gè)線程池，用于管理多個(gè)線程的執(zhí)行，而 RecursiveTask 類則是一個(gè)抽象類，用于定義可分解的任務(wù)。通過使用這些類，開發(fā)者可以非常方便地實(shí)現(xiàn) Fork/Join 的并行計(jì)算功能，從而提高應(yīng)用程序的性能和效率。

總之，F(xiàn)ork/Join 并不是一個(gè)框架，而是一種并發(fā)編程技術(shù)，它可以幫助開發(fā)者實(shí)現(xiàn)高效的并行計(jì)算，并發(fā)揮多核 CPU 的計(jì)算能力。

??Forkjoin的方法

Fork/Join 框架提供了一些核心的方法來支持任務(wù)的分解和合并，下面我會(huì)對這些方法進(jìn)行理論講解：

• fork() 方法：fork() 方法用于將當(dāng)前任務(wù)進(jìn)行分解，將其拆分成更小的子任務(wù)并提交給 Fork/Join 框架進(jìn)行并行處理。該方法會(huì)異步地啟動(dòng)一個(gè)新的子任務(wù)，并立即返回，不會(huì)阻塞當(dāng)前線程。
• join() 方法：join() 方法用于等待子任務(wù)的執(zhí)行完成，并獲取其結(jié)果。在調(diào)用 join() 方法之前，程序會(huì)阻塞當(dāng)前線程，直到子任務(wù)的執(zhí)行完成。如果子任務(wù)還沒有完成，則當(dāng)前線程會(huì)暫停執(zhí)行，轉(zhuǎn)而執(zhí)行其他可執(zhí)行任務(wù)，從而實(shí)現(xiàn)工作竊取的效果。
• invoke() 方法：invoke() 方法用于提交一個(gè)任務(wù)給 Fork/Join 框架進(jìn)行處理，并返回任務(wù)的執(zhí)行結(jié)果。該方法會(huì)同步地啟動(dòng)一個(gè)任務(wù)，并阻塞當(dāng)前線程，直到任務(wù)執(zhí)行完成并返回結(jié)果。這個(gè)方法通常用于提交根任務(wù)或最頂層的任務(wù)。
• RecursiveTask 類和 RecursiveAction 類：Fork/Join 框架提供了兩個(gè)抽象類 RecursiveTask 和 RecursiveAction，用于定義可分解的任務(wù)。RecursiveTask 適用于需要返回結(jié)果的任務(wù)，而 RecursiveAction 適用于不需要返回結(jié)果的任務(wù)。這兩個(gè)類都有一個(gè)抽象方法 compute()，我們需要在子類中實(shí)現(xiàn)該方法來定義具體的任務(wù)邏輯。
• Work-Stealing（工作竊?。篎ork/Join 框架采用了工作竊取算法，使得線程在處理完自己的任務(wù)后可以從其他線程的隊(duì)列中偷取任務(wù)來執(zhí)行。這種方式可以提高并行計(jì)算的效率和負(fù)載均衡性能。工作竊取是通過雙端隊(duì)列實(shí)現(xiàn)的，每個(gè)線程都有自己的任務(wù)隊(duì)列，當(dāng)一個(gè)線程完成自己隊(duì)列中的任務(wù)后，它會(huì)嘗試從其他線程的隊(duì)列末尾竊取任務(wù)來執(zhí)行。

這些方法和概念是 Fork/Join 框架中非常重要的部分，它們通過任務(wù)的分解、合并和工作竊取機(jī)制，實(shí)現(xiàn)了高效的并行計(jì)算。理解并熟練使用這些方法可以幫助開發(fā)者更好地利用 Fork/Join 框架來處理并行計(jì)算任務(wù)。

??代碼實(shí)現(xiàn)

ForkjoinDemo.java

package org.Test6;

import java.util.concurrent.RecursiveTask;

public class ForkjoinDemo extends RecursiveTask<Long> {
    private Long start; // 1
    private Long end; // 1990900000
    // 臨界值
    private Long temp = 10000L;

    public ForkjoinDemo(Long start, Long end) {
        this.start = start;
        this.end = end;
    }

    // 計(jì)算方法
    @Override
    protected Long compute() {
        if ((end - start) < temp) {
            Long sum = 0L;
            for (Long i = start; i <= end; i++) {
                sum += i;
            }
            return sum;
        } else { // forkjoin 遞歸
            long middle = (start + end) / 2; // 中間值
            ForkjoinDemo task1 = new ForkjoinDemo(start, middle);
            task1.fork(); // 拆分任務(wù)，把任務(wù)壓入線程隊(duì)列
            ForkjoinDemo task2 = new ForkjoinDemo(middle + 1, end);
            task2.fork(); // 拆分任務(wù)，把任務(wù)壓入線程隊(duì)列
            return task1.join() + task2.join();
        }
    }
}

Test.java

package org.Test6;

import java.util.concurrent.ExecutionException;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.ForkJoinTask;
import java.util.concurrent.RecursiveTask;
import java.util.stream.LongStream;

public class Test {

    public static void test1() {
        Long sum = 0L;
        long start = System.currentTimeMillis();
        for (Long i = 1L; i <= 10_0000_0000; i++) {
            sum += i;
        }
        long end = System.currentTimeMillis();
        System.out.println("sum=" + sum + " 時(shí)間：" + (end - start));
    }

    // 會(huì)使用ForkJoin
    public static void test2() throws ExecutionException, InterruptedException {
        long start = System.currentTimeMillis();
        ForkJoinPool forkJoinPool = new ForkJoinPool();
        ForkJoinTask<Long> task = new ForkjoinDemo(0L, 10_0000_0000L);
        ForkJoinTask<Long> submit = forkJoinPool.submit(task);// 提交任務(wù)
        Long sum = submit.get();
        long end = System.currentTimeMillis();
        System.out.println("sum=" + sum + " 時(shí)間：" + (end - start));
    }

    public static void test3() {
        long start = System.currentTimeMillis();
        // Stream并行流
        long sum = LongStream.rangeClosed(0L, 10_0000_0000L).parallel().reduce(0, Long::sum);
        long end = System.currentTimeMillis();

        System.out.println("sum=" + sum + " 時(shí)間：" + (end - start));
    }

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        test1();
        test2();
        test3();
    }

}

在技術(shù)的道路上，我們不斷探索、不斷前行，不斷面對挑戰(zhàn)、不斷突破自我?？萍嫉陌l(fā)展改變著世界，而我們作為技術(shù)人員，也在這個(gè)過程中書寫著自己的篇章。讓我們攜手并進(jìn)，共同努力，開創(chuàng)美好的未來！愿我們在科技的征途上不斷奮進(jìn)，創(chuàng)造出更加美好、更加智能的明天！

文章來源地址http://www.zghlxwxcb.cn/news/detail-795993.html

到了這里，關(guān)于【并發(fā)編程篇】詳解Forkjoin的文章就介紹完了。如果您還想了解更多內(nèi)容，請?jiān)谟疑辖撬阉鱐OY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！