ForkJoin
概念
Fork/Join是JDK 1.7加入的新的线程池实现，它体现的是一种分治思想，适用于能够进行任务拆分的cpu密集型运算
所谓的任务拆分，是将一个大任务拆分为算法上相同的小任务，直至不能拆分可以直接求解。使用ForkJoin框架需要将逻辑操作进行分治. 比较适合可以使用动态规划、回溯、递归等算法实现的逻辑。比如快速排序、二分查找，可以使用ForkJoin来提高运算能力。
Fork/Join在分治的基础上加入了多线程，可以把每个任务的分解和合并交给不同的线程来完成，进一步提升了运算效率
基本用法
- 定义任务类：创建一个继承自
RecursiveTask或RecursiveAction的任务类。RecursiveTask用于有返回结果的任务，而RecursiveAction用于没有返回结果的任务。 - 实现compute()方法：在任务类中重写
compute()方法，定义任务的具体计算逻辑。在该方法中，可以进行任务的拆分、执行以及结果的合并。 - 创建任务对象和ForkJoinPool：在主程序中，创建任务对象并将其提交给
ForkJoinPool进行执行。
案例代码：
java">import java.util.concurrent.ExecutionException;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.ForkJoinTask;
import java.util.concurrent.RecursiveTask;
import java.util.logging.Logger;
/**
* 计算从begin到end的和
*/
public class MyRecursivelyTask extends RecursiveTask<Integer> {
private static Logger log = Logger.getLogger("zzz");
private Integer begin;
private Integer end;
public MyRecursivelyTask(Integer begin, Integer end) {
this.begin = begin;
this.end = end;
}
/**
* 实现compute方法,拆分任务执行
**/
@Override
protected Integer compute() {
if (begin.equals(end)) {
log.info("begin:"+ begin + "end:"+ end);
return begin;
}
if (end - begin == 1) {
log.info("begin:"+ begin + "end:"+ end);
return begin + end;
}
Integer mid = (begin+end)/2;
// 拆分任务
MyRecursivelyTask t1 = new MyRecursivelyTask(mid + 1, end);
MyRecursivelyTask t2 = new MyRecursivelyTask(begin, mid);
log.info("t1:"+t1.toString());
log.info("t2:"+t2.toString());
// 执行拆分任务
t1.fork();
t2.fork();
// 获取拆分的任务的结果
Integer res1 = t1.join();
Integer res2 = t2.join();
return res1 + res2;
}
@Override
public String toString() {
return "MyRecursivelyTask{" +
"begin=" + begin +
", end=" + end +
'}';
}
}
class forkJoinTest{
public static void main(String[] args) throws ExecutionException, InterruptedException {
// 创建任务, 计算从1到5的和
MyRecursivelyTask t = new MyRecursivelyTask(1, 5);
// 创建fork/join线程池设置并行度为4
ForkJoinPool pool = new ForkJoinPool(4);
// 提交任务
ForkJoinTask<Integer> res = pool.submit(t);
// 获取任务结果
System.out.println(res.get());
}
}
ForkJoinPool中的API
Future<T> submit(ForkJoinTask<T> task)：- 提交一个ForkJoinTask任务进行执行，并返回一个Future对象，可通过该对象获取任务的执行结果。
T invoke(ForkJoinTask<T> task)：- 提交一个ForkJoinTask任务并立即执行，阻塞调用线程，直到任务完成并返回结果。
void execute(ForkJoinTask<?> task)：- 提交一个ForkJoinTask任务进行执行，但不返回任何结果。该方法不会阻塞调用线程。
void shutdown()：- 优雅地关闭ForkJoinPool，停止接受新任务，等待已提交的任务完成执行。
List<Runnable> shutdownNow()：- 立即关闭ForkJoinPool，试图中断正在执行的任务并返回所有未执行的任务列表。
boolean awaitTermination(long timeout, TimeUnit unit)：- 阻塞调用线程，等待指定的时间，直到所有任务完成执行或超时。
int getPoolSize()：- 获取当前ForkJoinPool中的工作线程数。
int getActiveThreadCount()：- 获取当前正在执行任务的工作线程数。
int getParallelism()：- 获取ForkJoinPool的并行度，即最大并行执行的线程数。
long getStealCount()：- 获取ForkJoinPool中的工作线程从其他工作线程中偷取任务的次数。
long getQueuedTaskCount()：- 获取当前在ForkJoinPool队列中等待执行的任务数。
