14、分支合并：ForkJoin
14.1. 什么是ForkJoin
ForkJoin,任务切分、合并操作。大数据中的mapreduce ，就是任务切分，结果合并。原理如下图所示：
14.2. 工作窃取
工作窃取
A任务1 –> 任务2 –> 任务3 –> 任务4 A领先执行完成 ，帮B执行任务（从B任务的尾部开始窃取任务执行）！
B任务1 –> 任务2 任务3 任务4
工作开始从头，窃取从尾，有效提高速度，双端队列
14.3. 核心类
14.3.1. ForkJoinPool
1、ForkJoinPool
ForkJoinPool ，是实现了ExecutorService的任务池，将任务ForkJoinTask放到ForkJoinPool中，去运行线程。 通过队列来执行，找到实现接口的类
而ForkJoinPool 中存在一个内部类,工作队列WorkQueue,是ForkJoinPool 的 一个内部类。每一个线程都有一个 WorkQueue ！
14.3.2. ForkJoinTask
2、ForkJoinTask
ForkJoinTask 是一个抽象类，代表正在 ForkJoinPool 中运行的 任务，它有三个主要的方法：
fork： 安排任务异步执行，简单的说，就是创建一个子任务。
join：当任务完成后获取去返回的计算结果！
invoke：开始执行！如果计算没有完毕，就会等待！
14.3.3. RecursiveTask
3、RecursiveTask
ForkJoinTask的一个重要子类 递归 RecursiveTask
其中有个计算方法compute
/**
* The main computation performed by this task.
* @return the result of the computation
*/
protected abstract V compute();
我们一般要继承RecursiveTask类，重写compute方法，如下示例：
MyRecursiveTask
package com.interview.concurrent.stream;
import java.util.concurrent.RecursiveTask;
/**
* @author DDKK.COM 弟弟快看，程序员编程资料站
* @description 描述：递归任务，泛型是计算后返回的结果类型
* @date 2023/2/24 15:56
*/
public class MyRecursiveTask extends RecursiveTask<Long> {
private long start; //开始值
private long end; //结束值
private static final long temp = 10000L; //中间值
public MyRecursiveTask(long start, long end) {
this.start = start;
this.end = end;
}
@Override
protected Long compute() {
if (end - start <= temp) {
long sum = 0L;
for (long i = start; i <= end; i++) {
sum += i;
}
return sum;
}else{
//获取中间值
long middle = (start + end)/2;
/**
* fork()会不断的循环
*/
//第一个任务
MyRecursiveTask rightTask = new MyRecursiveTask(start,middle);
rightTask.fork();
//第二个任务
MyRecursiveTask leftTask = new MyRecursiveTask(middle+1, end);
leftTask.fork();
//合并结果
return rightTask.join() + leftTask.join();
}
}
}
fork()会不断的循环。
ForkJoin代码编写模型：
1、创建ForkJoinPool;
2、创建ForkJoinTask;
3、ForkJoinPool对象调用invoke执行，并将ForkJoinTask对象放入ForkJoinPool中
示例：计算1到10,0000,0000的和，通过三种方式，比较性能
package com.interview.concurrent.stream;
import java.util.concurrent.ForkJoinPool;
import java.util.stream.LongStream;
/**
* @author DDKK.COM 弟弟快看，程序员编程资料站
* @description 描述:计算1到10,0000,0000的和
* @date 2023/2/24 15:40
*/
public class ForkJoinDemo {
public static void main(String[] args) {
//calculateNormal(); //time:781 sum:500000000500000000
//calculateForkJoin(); //time:724 sum:500000000500000000
calculateStream(); //time:473 sum:500000000500000000
}
// 正常测试
public static void calculateNormal(){
long startTime = System.currentTimeMillis();
long sum = 0L;
for (long i = 0L; i <= 10_0000_0000L; i++) {
sum += i;
}
long endTime = System.currentTimeMillis();
System.out.println("time:"+(endTime-startTime)+" sum:"+sum);
}
// ForkJoin测试
public static void calculateForkJoin(){
long startTime = System.currentTimeMillis();
/**
*1、创建ForkJoinPool;
*/
ForkJoinPool forkJoinPool = new ForkJoinPool();
/**
*2、创建ForkJoinTask;
*/
MyRecursiveTask recursiveTask = new MyRecursiveTask(0L,10_0000_0000L);
/**
3、ForkJoinPool对象调用invoke执行，并将ForkJoinTask对象放入ForkJoinPool中;
*/
long sum = forkJoinPool.invoke(recursiveTask);
long endTime = System.currentTimeMillis();
System.out.println("time:"+(endTime-startTime)+" sum:"+sum);
}
// Stream并行流测试
public static void calculateStream(){
long startTime = System.currentTimeMillis();
long sum = LongStream.rangeClosed(0L,10_0000_0000L).parallel().reduce(0L,Long::sum);
long endTime = System.currentTimeMillis();
System.out.println("time:"+(endTime-startTime)+" sum:"+sum);
}
}