问题

RecursiveAction是什么?RecursiveTask是什么?

答案

RecursiveAction

在下面的例子中,将对一个字符串进行处理,我们简单的将字符串转成大写,并打印它.

为了证明Fork/Join框架的Fork(任务分解)行为,在字符串的长度大于我们指定的临界值的时候，我们会使用createSubtasks()方法,将字符串转成大写这个任务拆分成更小的任务,如果字符串小于或等于我们的临界值,将会执行processing()方法,将字符串转成大小并输出.

这个字符串被递归的分解成子字符串,基于这些子字符串再创建CustomRecursiveAction实例.

createSubtasks()方法返回一个CustomRecursiveAction列表.

通过调用ForkJoinTask.invokeAll()方法,这些任务列表将会提交给ForkJoinPool

public class CustomRecursiveAction extends RecursiveAction {

    private String workload = "";
    private static final int THRESHOLD = 4;

    public CustomRecursiveAction(String workload) {
        this.workload = workload;
    }

    @Override
    protected void compute() {
        if (workload.length() > THRESHOLD) {
            ForkJoinTask.invokeAll(createSubtasks());
        } else {
            processing(workload);
        }
    }

    private List<CustomRecursiveAction> createSubtasks() {
        List<CustomRecursiveAction> subtasks = new ArrayList<>();
        String partOne = workload.substring(0, workload.length() / 2);
        String partTow = workload.substring(workload.length() / 2, workload.length());
        subtasks.add(new CustomRecursiveAction(partOne));
        subtasks.add(new CustomRecursiveAction(partTow));
        return subtasks;
    }

    private void processing(String work) {
        String result = work.toUpperCase();
        System.out.println("结果 - (" + result + ") - 被" + Thread.currentThread().getName() + "处理");
    }

    public static void main(String[] args) {
        CustomRecursiveAction action = new CustomRecursiveAction("ABCDEFGHIJKLMN");
        action.compute();
    }

以上的代码思路为:

1. 创建一个对象代表所有的工作量.
2. 定义一个临界值,当工作量没到达这个临界值时,继续将任务进行分解.
3. 当任务达到临界值,或者在临界值之下时,开始执行任务.

基于这种思路,我们可以用来解决特定的问题.

RecursiveTask

RecusiveTask和RecusiveAction相似,只不过每个子任务处理之后会带一个返回值,最终所有的子任务的返回结果会join(合并)成一个结果.

public class CustomRecursiveTask extends RecursiveTask<Integer> {

    private int[] arr;
    private static final int THRESHOLD = 2;

    public CustomRecursiveTask(int[] arr) {
        this.arr = arr;
    }


    @Override
    protected Integer compute() {
        if (arr.length > THRESHOLD) {
            return ForkJoinTask.invokeAll(createSubtasks())
                    .stream()
                    .mapToInt(ForkJoinTask::join)
                    .sum();
        } else {
            return processing(arr);
        }
    }

    private List<CustomRecursiveTask> createSubtasks() {
        List<CustomRecursiveTask> subTasks = new ArrayList<>();
        subTasks.add(new CustomRecursiveTask(Arrays.copyOfRange(arr, 0, arr.length / 2)));
        subTasks.add(new CustomRecursiveTask(Arrays.copyOfRange(arr, arr.length / 2, arr.length)));
        return subTasks;
    }

    private Integer processing(int[] arr) {
        return Arrays.stream(arr)
                .filter(a -> a > 0 && a < 11)
                .map(a -> a * 10)
                .sum();
    }

    public static void main(String[] args) {
        int[] arr = {1, 2, 3, 4, 5, 6, 7, 8, 9, 10};
        CustomRecursiveTask customRecursiveTask = new CustomRecursiveTask(arr);
        Integer result = customRecursiveTask.compute();
        System.out.println("result:" + result);
    }
}

在这个例子中,我们处理的工作量是一个整型数组.在这里,invokeAll()方法提交子任务,并返回一个Future列表.通过Java8的Stream API的sum()方法,最终把任务的执行结果都相加(join)起来.