Fork/Join框架-CFANZ编程社区

Fork/Join框架是Java 7提供的一个用于并行执行任务的框架，是一个把大任务分割成若干个小任务(Fork)，最终汇总每个小任务结果后得到大任务结果(Join)的框架。

分而治之

一个规模为N的问题，当N小于阀值时直接执行，当N大于阀值时将N分解成K个小规模子问题，子问题之间相互独立，并与原问题形式相同，最后将所有子问题的解合并得到原问题的解，叫做分而治之。

Fork/Join框架_Fork/Join

工作窃取算法

工作窃取（work-stealing）算法是指某个线程从其他队列里窃取任务来执行。一般任务被拆分后会分配到不同的队列，并为每个队列建立单独的线程来处理任务，线程和队列一一对应。但是有的线程会将自己的任务先处理完，这个时候先处理完的线程就可以去帮助其他线程干活，以此来提升整个任务的工作效率。为了减少窃取任务时候的冲突，通常会使用双端队列，窃取任务的线程永远从队列尾部拿任务，正常线程从队列头部拿任务。

Fork/Join框架_Fork/Join实战_02

工作窃取算法的优点：充分利用线程进行并行计算，减少了线程间的竞争。
工作窃取算法的缺点：在某些情况下还是存在竞争，比如双端队列里只有一个任务时。并且该算法会消耗了更多的系统资源，比如创建多个线程和多个双端队列。

Fork/Join框架使用

使用的标准范式

Fork/Join框架_子任务_03

invokeAll()方法里面最终调用的还是 fork()方法。
ForkJoinPool新建的线程数默认等于CPU核数

Fork/Join使用两个类来完成任务分割和执行任务合并结果两件事情。

ForkJoinTask：我们要使用ForkJoin框架，必须首先创建一个ForkJoin任务。它提供在任务中执行fork()和join()操作的机制。通常情况下，我们不需要直接继承ForkJoinTask类，只需要继承它的子类，Fork/Join框架提供了以下两个子类。

RecursiveAction：用于没有返回结果的任务。
RecursiveTask：用于有返回结果的任务。

ForkJoinPool：ForkJoinTask需要通过ForkJoinPool来执行。
任务分割出的子任务会添加到当前工作线程所维护的双端队列中，进入队列的头部。当一个工作线程的队列里暂时没有任务时，它会随机从其他工作线程的队列的尾部获取一个任务。

示例

计算1+2+3+4+…+n的值

package com.xiaolyuh;

import java.util.Objects;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.RecursiveTask;

/**
 * 计算1+2+3+...+n的值
 * 使用同步执行的方式
 *
 * @author yuhao.wang3
 * @since 2019/6/25 17:07
 */
public class ForkJoinCountTask extends RecursiveTask<Long> {

    /**
     * 阀值
     */
    private int threshold = 10;

    /**
     * 任务的开始值
     */
    private long start;

    /**
     * 任务的结束值
     */
    private long end;

    public ForkJoinCountTask(long start, long end) {
        this.start = start;
        this.end = end;
    }

    @Override
    protected Long compute() {
        if (end - start <= threshold) {
            long count = 0;
            for (int i = 0; i <= end - start; i++) {
                count = count + start + i;
            }
            return count;
        } else {
            // 如果任务大于阈值，就分裂成三个子任务计算
            long slip = (end - start) / 3;
            ForkJoinCountTask oneTask = new ForkJoinCountTask(start, start + slip);
            ForkJoinCountTask twoTask = new ForkJoinCountTask(start + slip + 1, start + slip * 2);
            ForkJoinCountTask threeTask = new ForkJoinCountTask(start + slip * 2 + 1, end);
            // 提交子任务到框架去执行
            invokeAll(oneTask, twoTask, threeTask);
            // 等待子任务执行完，得到其结果，并合并子任务
            return oneTask.join() + twoTask.join() + threeTask.join();
        }
    }

    public static void main(String[] args) {
        long start = System.currentTimeMillis();
        ForkJoinPool pool = new ForkJoinPool();
        // 生成一个计算任务，负责计算1+2+3+n
        ForkJoinCountTask countTask = new ForkJoinCountTask(1, 1000000);
        // 执行一个任务（同步执行，任务会阻塞在这里直到任务执行完成）
        pool.invoke(countTask);
        // 异常检查
        if (countTask.isCompletedAbnormally()) {
            Throwable throwable = countTask.getException();
            if (Objects.nonNull(throwable)) {
                System.out.println(throwable.getMessage());
            }
        }
        // join方法是一个阻塞方法，会等待任务执行完成
        System.out.println("计算为：" + countTask.join() + ", 耗时：" + (System.currentTimeMillis() - start) + "毫秒");
    }
}

搜索指定目录下的指定文件

package com.xiaolyuh;

import java.io.File;
import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.RecursiveAction;

/**
 * 搜索指定目录下的指定文件
 * 使用异步执行的方式
 *
 * @author yuhao.wang3
 * @since 2019/6/25 17:07
 */
public class ForkJoinSearchFileTask extends RecursiveAction {

    /**
     * 指定目录
     */
    private File file;

    /**
     * 文件后缀
     */
    private String suffix;

    public ForkJoinSearchFileTask(File file, String suffix) {
        this.file = file;
        this.suffix = suffix;
    }

    @Override
    protected void compute() {
        if (Objects.isNull(file)) {
            return;
        }

        File[] files = file.listFiles();
        List<ForkJoinSearchFileTask> fileTasks = new ArrayList<>();
        if (Objects.nonNull(files)) {
            for (File f : files) {
                // 拆分任务
                if (f.isDirectory()) {
                    fileTasks.add(new ForkJoinSearchFileTask(f, suffix));
                } else {
                    if (f.getAbsolutePath().endsWith(suffix)) {
                        System.out.println("文件： " + f.getAbsolutePath());
                    }
                }
            }
            // 提交并执行任务
            invokeAll(fileTasks);
            for (ForkJoinSearchFileTask fileTask : fileTasks) {
                // 等待任务执行完成
                fileTask.join();
            }
        }
    }

    public static void main(String[] args) throws Exception {
        File file = new File("d:/");
        ForkJoinPool pool = new ForkJoinPool();
        // 生成一个计算任务，负责查找指定木目录
        ForkJoinSearchFileTask searchFileTask = new ForkJoinSearchFileTask(file, ".txt");
        // 异步执行一个任务
        pool.execute(searchFileTask);

        Thread.sleep(10);

        // 做另外的事情
        int count = 0;
        for (int i = 0; i < 1000; i++) {
            count += i;
        }
        System.out.println("计算任务：" + count);

        // join方法是一个阻塞方法，会等待任务执行完成
        searchFileTask.join();
    }
}

Fork/Join框架的异常处理

ForkJoinTask在执行的时候可能会抛出异常，但是我们没办法在主线程里直接捕获异常，所以ForkJoinTask提供了isCompletedAbnormally()方法来检查任务是否已经抛出异常或已经被取消了，并且可以通过ForkJoinTask的getException方法获取异常。getException方法返回Throwable对象，如果任务被取消了则返回CancellationException。如果任务没有完成或者没有抛出异常则返回null。

// 异常检查
if(countTask.isCompletedAbnormally()) {
    Throwable throwable = countTask.getException();
    if (Objects.nonNull(throwable)) {
        System.out.println(throwable.getMessage());
    }
}