线程并发工具类之Fork-Join框架

Fork-Join框架

分治法

分治算法的基本思想将一个规模为N的问题，分解成K个规模较小的子问题，这些子问题相互独立且月原问题性质相同。求解出子问题的解，合并得到原问题的解。

Fork-Join框架

任务分隔：体现了分而治之的思维，将一个大任务，进行拆分(Fork)成分成若干个小任务（拆到不能拆分为止）
执行任务并合并结果：分割的子任务分别放在双端队列中，然后启动多个线程分别从双端队列中获取子任务并执行，子任务完成后的结果放到另外一个里，启动一个线程从这个队列里取数据计算后返回

image.png

Fork-Join框架使用标准范式

• ForkJoinTask：创建ForkJoinTask任务，该类提供了fork()、join()机制，通常继承它的RecursiveTask、RecursiveAction抽象类，RecursiveTask用于返回值任务，RecursiveAction用于无返回值任务
• ForkJoinPool：创建ForkJoinPool对象，用于执行ForkJoinTask任务
• 拆分任务并提交到上一级Task
• ForkJoinPool池invoke任务
• join所有的之任务结果

工作密取

多线程操作双端队列，每个消费者都有各自的双端队列。如果一个消费者完成了自己双端队列中的全部工作，那么它可以从其它消费者双端队列末尾秘密地获取工作。

任务分割后的子任务会添加到当前线程所维护的双端队列中，进入队列头部，当一个工作线程的队列任务暂时没有时，会随机从其他工作线程的双端队列尾部获取一个任务执行

Fork-Join框架同步返回计算数组求和

import java.util.concurrent.RecursiveTask;

import com.shawntime.enjoy.architect.concurrency.SleepUtils;

public class ArraySumTask extends RecursiveTask<Long> {

    private int[] array;

    private int startIndex;

    private int endIndex;

    private int minNum;

    public ArraySumTask(int[] array, int startIndex, int endIndex, int minNum) {
        this.array = array;
        this.startIndex = startIndex;
        this.endIndex = endIndex;
        this.minNum = minNum;
    }

    @Override
    protected Long compute() {
        if (endIndex - startIndex < minNum) {
            // 已经是最小了
            Long result = 0L;
            for (int i = startIndex; i <= endIndex; ++i) {
                SleepUtils.sleepByMilliSeconds(1);
                result += array[i];
            }
            return result;
        } else {
            System.out.println("继续拆分....");
            // 继续拆分
            int middle = (endIndex + startIndex) / 2;
            ArraySumTask leftTask = new ArraySumTask(array, startIndex, middle, minNum);
            ArraySumTask rightTask = new ArraySumTask(array, middle + 1, endIndex, minNum);
            // 提交任务
            invokeAll(leftTask, rightTask);
            return leftTask.join() + rightTask.join();
        }
    }
}

int[] array = ArrayUtils.makeArray(arrayLength);
ForkJoinPool forkJoinPool = new ForkJoinPool();
int minNum = arrayLength / 50;
ForkJoinTask<Long> task = new ArraySumTask(array, 0, array.length - 1, minNum);
forkJoinPool.invoke(task);
Long value = task.join();

Fork-Join框架异步遍历查找文件

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

/**
 * 异步的方式不带返回值
 */
public class FindFileTask extends RecursiveAction {

    private File file;

    public FindFileTask(File file) {
        this.file = file;
    }

    @Override
    protected void compute() {
        if (file == null) {
            return;
        }
        File[] files = file.listFiles();
        List<FindFileTask> tasks = new ArrayList<>();
        for (File subFile : files) {
            if (subFile.isDirectory()) {
                FindFileTask task = new FindFileTask(subFile);
                tasks.add(task);
            } else {
                if (subFile.getName().endsWith("mp4")) {
                    System.out.println(subFile.getAbsolutePath());
                }
            }
        }
        if (tasks.isEmpty()) {
            return;
        }
        for (FindFileTask task : invokeAll(tasks)) {
            task.join();
        }
    }
}

public static void main(String[] args) {
    ForkJoinPool forkJoinPool = new ForkJoinPool();
    File file = new File("D:/");
    FindFileTask findFiles = new FindFileTask(file);
    // 异步提交
    forkJoinPool.execute(findFiles);

    int result = 0;
    for (int i = 0; i < 1000; ++i) {
        result += i;
    }

    System.out.println("main result : " + result);

    findFiles.join(); // 阻塞

    System.out.println("end....");
}