概要
现代的计算机已经向多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 目标和是:412 目标和是:616 目标和是:1065 目标和是:1452 目标和是:1697 目标和是:1801 目标和是:1874 目标和是:2362 目标和是:2404 目标和是:2636 目标和是:2720 目标和是:3140 目标和是:3241 目标和是:3666 目标和是:3669 目标和是:4151 目标和是:4159 目标和是:4422 目标和是:4914 目标和是:5221 目标和是:5533 目标和是:5971 目标和是:6000 目标和是:6152 目标和是:6619 目标和是:6732 目标和是:6997 目标和是:7069 目标和是:7498 目标和是:7939 目标和是:8138 目标和是:8389 目标和是:8805 目标和是:9148 目标和是:9534 目标和是:9582 目标和是:9985 目标和是:10277 目标和是:10509 目标和是:10921 目标和是:11390 目标和是:11888 目标和是:12027 目标和是:12164 目标和是:12345 目标和是:12769 目标和是:12821 目标和是:13289 目标和是:13549 目标和是:13599 目标和是:13763 目标和是:13835 目标和是:14094 目标和是:14333 目标和是:14781 目标和是:15021 目标和是:15436 目标和是:15473 目标和是:15659 目标和是:15793 目标和是:15940 目标和是:16272 目标和是:16444 目标和是:16552 目标和是:16757 目标和是:16948 目标和是:17142 目标和是:17196 目标和是:17555 目标和是:17896 目标和是:18244 目标和是:18358 目标和是:18763 目标和是:19059 目标和是:19073 目标和是:19495 目标和是:19770 目标和是:20070 目标和是:20483 目标和是:20757 目标和是:21036 目标和是:21490 目标和是:21703 目标和是:22013 目标和是:22109 目标和是:22598 目标和是:22694 目标和是:22961 目标和是:23211 目标和是:23324 目标和是:23576 目标和是:23901 目标和是:24064 目标和是:24369 目标和是:24575 目标和是:24857 目标和是:25002 目标和是:25491 目标和是: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”框架。