ForkJoin 应用示例

1. 前言

上一节我们学习了 ForkJoin 的基本概念和核心 API,本节带领大家实现一个具体的应用案例。从实际应用中感受一下 ForkJoin 框架的使用,以及此框架带来的便利。

本节先描述待实现的案例内容,接着做编码实现,然后总结使用过程中的注意事项。

2. 案例描述

我们在实际项目中,常常会碰到大数据集合的处理,如大文件、大表、内存中的大数据集合。由于数据量大,我们常常会考虑采用多线程的思路提高处理效率。
当待处理的数据集的每一部分的数据处理逻辑基本一致,且可以很好拆分成小的数据集进行处理时,使用 ForkJoin 进行处理比好合适。

我们还是采用 Executor 应用示例中的场景:需要对某个目录下的所有文件(成百上千)进行加密并用文件的 MD5 串修改文件名称。

在开始动手实现之前,我们先做一个简单的分析。在这个案例中,我们将 “对文件进行加密、生成 MD5 串、修改文件名称” 作为待执行任务的内容。所有文件形成的列表就是我们待处理的数据范围。为了校验整个处理过程是否有文件遗漏,我们最终需要核对处理结果,所以我们用 FileForkJoinTask 类(继承 RecursiveTask)封装我们的任务逻辑。为了方便演示,下面编码中部分数据采用了模拟的方式生成。

3. 编码实现

import java.util.Random;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.Future;

public class ForkJoinTest {

    // 模拟待处理的文件列表
    private static int fileListSize = new Random().nextInt(15);
    private static String[] fileList = new String[fileListSize];
    static {
        for(int i=0; i<fileListSize; i++) {
            fileList[i] = "fileName" + i;
        }
    }

    // 主线程
    public static void main(String[] args) throws Exception {
        // 创建用于处理任务的线程池
        // ForkJoinPool forkJoinPool = ForkJoinPool.commonPool(); 这种创建方式可最大化使用全局系统资源
        ForkJoinPool forkJoinPool = new ForkJoinPool();
        // 提交待处理的总任务
        Future result = forkJoinPool.submit(new FileDealTask(0, fileListSize, fileList));
        // 获取任务执行结果
        System.out.println("预备处理的文件个数" + fileListSize + ",总共处理的文件个数:" + result.get());
        // 关闭线程池
        forkJoinPool.shutdown();
    }
}

上面代码注释已经很清楚了,我们观察下面的代码,看看任务是怎么切分的,以及子任务的结果是怎么做汇总的。

import lombok.SneakyThrows;
import java.util.Random;
import java.util.concurrent.RecursiveTask;

public class FileDealTask extends RecursiveTask<Integer> {

    private String[] fileList;
    // 当子任务划分到只需要处理最多10个文件时,停止分割任务
    private final int threshold = 2;
    private int first;
    private int last;

    public FileDealTask(int first, int last, String[] fileList) {
        this.fileList = fileList;
        this.first = first;
        this.last = last;
    }

    @SneakyThrows
    @Override
    protected Integer compute() {
        // 执行结果
        int result = 0;

        // 任务足够小则直接处理(对文件进行加密、生成MD5串、修改文件名称)
        if (last - first <= threshold) {
            for (int i = first; i < last; i++) {
                result = result + 1;
                Thread.sleep(new Random().nextInt(2000));
                System.out.println(Thread.currentThread().getName() + ":文件" + fileList[i] + "已处理完毕");
            }
            System.out.println(Thread.currentThread().getName() + ":总共处理的文件数 (" + first + "," + last + ")" + result);
        } else {
            // 拆分成小任务
            int middle = first + (last - first) / 2;
            // 创建两个子任务
            FileDealTask leftTask = new FileDealTask(first, middle, fileList);
            FileDealTask rightTask = new FileDealTask(middle, last, fileList);
            // 触发两个子任务开始执行
            invokeAll(leftTask, rightTask);
            // 等待两个子任务执行结果并返回
            result = leftTask.join() + rightTask.join();
            System.out.println(Thread.currentThread().getName() + ":当前任务继续拆分 "
                    + " (" + first + "," + middle + "), (" + (middle) + "," + last + ")");
        }
        return result;
    }
}

我们通过在 IDE 中运行上面这个示例,看看实际的运行结果。
【补充视频】

上面代码逻辑中有随机内容,每次运行结果会有差异,运行上面的代码,我们观察某次运行结果如下:

ForkJoinPool-1-worker-2:文件fileName3已处理完毕
ForkJoinPool-1-worker-2:总共处理的文件数 (3,4)1
ForkJoinPool-1-worker-1:文件fileName0已处理完毕
ForkJoinPool-1-worker-1:总共处理的文件数 (0,1)1
ForkJoinPool-1-worker-0:文件fileName4已处理完毕
ForkJoinPool-1-worker-3:文件fileName1已处理完毕
ForkJoinPool-1-worker-3:文件fileName2已处理完毕
ForkJoinPool-1-worker-3:总共处理的文件数 (1,3)2
ForkJoinPool-1-worker-1:当前任务继续拆分  (0,1), (1,3)
ForkJoinPool-1-worker-0:文件fileName5已处理完毕
ForkJoinPool-1-worker-0:总共处理的文件数 (4,6)2
ForkJoinPool-1-worker-2:当前任务继续拆分  (3,4), (4,6)
ForkJoinPool-1-worker-1:当前任务继续拆分  (0,3), (3,6)
预备处理的文件个数6,总共处理的文件个数:6

首先做了 (0,3)~(3,6),之后对 (0,3) 做了 (0,1), (1,3) 的拆分,对 (3,6) 做了 (3,4), (4,6) 的拆分。和我们的预期一致。

4. 注意事项

  1. ForkJoinPool 不是为了替代 ExecutorService,其主要用于实现 “分而治之” 的算法,最适合处理计算密集型的任务。
  2. 做好评估权衡,当需要处理的数据量不是特别大时,没有必要使用 ForkJoin。其底层使用多线程的方式处理任务,涉及到线程上下文的切换,当数据量不大的时候使用串行会比使用多线程快。
  3. 执行子任务时候要注意,使用 invokeAll,不能分别对子任务调用 fork。

5. 视频演示

6. 小结

本节通过一个实际例子的编码实现,展示了 ForkJoinPool 的具体用法。当然本节中的用法相对比较简单,更多的用法还需要大家进一步学习,希望大家多思考勤练习,早日掌握之。