概要
现代的计算机已经向多CPU方向发展，即使是普通的PC，甚至现在的智能手机、多核处理器已被广泛应用。在未来，处理器的核心数将会发展的越来越多。
虽然硬件上的多核CPU已经十分成熟，但是很多应用程序并未这种多核CPU做好准备，因此并不能很好地利用多核CPU的性能优势。
为了充分利用多CPU、多核CPU的性能优势，级软基软件系统应该可以充分“挖掘”每个CPU的计算能力，决不能让某个CPU处于“空闲”状态。为此，可以考虑把一个任务拆分成多个“小任务”,把多个"小任务"放到多个处理器核心上并行执行。当多个“小任务”执行完成之后，再将这些执行结果合并起来即可。
Java在JDK7之后加入了并行计算的框架Fork/Join，可以解决我们系统中大数据计算的性能问题。Fork/Join采用的是分治法，Fork是将一个大任务拆分成若干个子任务，子任务分别去计算，而Join是获取到子任务的计算结果，然后合并，这个是递归的过程。子任务被分配到不同的核上执行时，效率最高。伪代码如下：
- Result solve(Problem problem) {
- if (problem is small)
- directly solve problem
- else {
- split problem into independent parts
- fork new subtasks to solve each part
- join all subtasks
- compose result from subresults
- }
- }
Fork/Join框架的核心类是ForkJoinPool，它能够接收一个ForkJoinTask，并得到计算结果。ForkJoinTask有两个子类，RecursiveTask（有返回值）和RecursiveAction（无返回结果），我们自己定义任务时，只需选择这两个类继承即可
package test_lock; import java.util.ArrayList; import java.util.List; import java.util.Map; import java.util.concurrent.RecursiveTask; public class SumTask extends RecursiveTask<Integer>{ private static final int THRESHOLD=20; private int[] array; private int low; private int high; public SumTask(int[] array_p,int low_p,int high_p){ this.array=array_p; this.low=low_p; this.high=high_p; } @Override protected Integer compute() { // TODO Auto-generated method stub int sum =0; if((high-low + 1)<=THRESHOLD){ System.out.println(low +"-"+high +" 计算"); for(int i=low;i<=high;i++){ sum+=array[i]; } }else{ System.out.println(low +"-"+high+" 切分"); //1. 一个大任务，分割成两个子任务； int mid=(low+high)/2; SumTask left =new SumTask(array,low,mid); SumTask right=new SumTask(array,mid+1,high); //2.分别进行计算 invokeAll(left,right); //3.合并结果 sum =left.join()+right.join(); //另一种方法 try{ sum=left.get()+right.get(); }catch(Throwable e){ System.out.println(e.getMessage()); } } return sum; } public static void main(String[] args) { // TODO Auto-generated method stub // 1.6 泛型实例的创建可以通过类型推断来简化 可以去掉后面new部分的泛型类型，只用<>就可以了。 //使用泛型前 List strList = new ArrayList(); List<String> strList4 = new ArrayList<String>(); List<Map<String, List<String>>> strList5 = new ArrayList<Map<String, List<String>>>(); //编译器使用尖括号 (<>) 推断类型 List<String> strList0 = new ArrayList<String>(); List<Map<String, List<String>>> strList1 = new ArrayList<Map<String, List<String>>>(); List<String> strList2 = new ArrayList<>(); List<Map<String, List<String>>> strList3 = new ArrayList<>(); List<String> list = new ArrayList<>(); list.add("A"); // The following statement should fail since addAll expects // Collection<? extends String> //list.addAll(new ArrayList<>()); List<String> strList7 = new ArrayList<String>(); for(int i=0;i<10;i++){ strList7.add(String.valueOf(i)); } List<String> ll=new ArrayList<String>(); List<Map<String,List<String>>> ll1=new ArrayList<>(); strList7.forEach(o->{System.out.println(o);}); } }
package test_lock; import java.util.Arrays; import java.util.Random; import java.util.concurrent.ExecutionException; import java.util.concurrent.ForkJoinPool; public class fork_join { public static void main(String[] args) throws InterruptedException, ExecutionException { // TODO Auto-generated method stub /** * 下面以一个有返回值的大任务为例，介绍一下RecursiveTask的用法。 大任务是：计算随机的100个数字的和。 小任务是：每次只能20个数值的和。 */ int[] array = genArray(); System.out.println(Arrays.toString(array)); int total=0; for(int i=0;i<array.length;i++){ System.out.println("目标和是："+total); total+=array[i]; } System.out.println("目标和是："+total); //1. 创建任务： SumTask sumTask=new SumTask(array,0,array.length-1); // 2。创建线程池，设置并行计算的个数 int processors=Runtime.getRuntime().availableProcessors(); System.out.println("processors="+processors); ForkJoinPool forkJoinPool =new ForkJoinPool(processors*2); // 3.提交任务到线程池 forkJoinPool.submit(sumTask); long begin=System.currentTimeMillis(); //4 获取结果 Integer result =sumTask.get();//wait for long end =System.currentTimeMillis(); System.out.println(String.format("结果 %s,耗时 %sms",result,end-begin)); if(result==total){ System.out.println("测试成功!!"); }else{ System.out.println("fork join 调用失败!!!"); } } private static int[] genArray() { // TODO Auto-generated method stub int[] array=new int[100]; for(int i=0;i<100;i++){ array[i]=new Random().nextInt(500); } return array; } }
结果为：
[412, 204, 449, 387, 245, 104, 73, 488, 42, 232, 84, 420, 101, 425, 3, 482, 8, 263, 492, 307, 312, 438, 29, 152, 467, 113, 265, 72, 429, 441, 199, 251, 416, 343, 386, 48, 403, 292, 232, 412, 469, 498, 139, 137, 181, 424, 52, 468, 260, 50, 164, 72, 259, 239, 448, 240, 415, 37, 186, 134, 147, 332, 172, 108, 205, 191, 194, 54, 359, 341, 348, 114, 405, 296, 14, 422, 275, 300, 413, 274, 279, 454, 213, 310, 96, 489, 96, 267, 250, 113, 252, 325, 163, 305, 206, 282, 145, 489, 253, 322] 目标和是：0.... 目标和是：25744 目标和是：26066 processors=4 0-99 切分 0-49 切分 0-24 切分 50-99 切分 75-99 切分 25-49 切分 25-37 计算 13-24 计算 0-12 计算 38-49 计算 88-99 计算 50-74 切分 63-74 计算 50-62 计算 75-87 计算 结果 26066,耗时 2ms 测试成功!!
上面的代码是一个100个整数累加的任务，切分到小于20个数的时候直接进行累加，不再切分。
我们通过调整阈值（THRESHOLD），可以发现耗时是不一样的。实际应用中，如果需要分割的任务大小是固定的，可以经过测试，得到最佳阈值；如果大小不是固定的，就需要设计一个可伸缩的算法，来动态计算出阈值。如果子任务很多，效率并不一定会高。
PS：类似的这种“分而治之”的需求场景，往往带有递归性，实际中，我们可以考虑任务是否具有“递归性”来决定是否使用“Fork-Join”框架。