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....");
}
