Java7之线程池ForkJoinPool

拼搏现实的明天。 2022-04-14 05:21 271阅读 0赞

\[b\]\[size=large\]许多情况下，在一个程序中使用多线程是有益处的，可以大大提高程序的效率，多线程主要有以下3个优点1,资源利用率更好2，程序设计在某些情况下更简单3，程序响应更快。当然凡事有利就有弊，多线程也会使程序的开发，维护及调试更加复杂，当然如果我们能够扬长避短，在正确的场合下使用多线程，那么它将成为我们程序中开发的利器。  
\[/size\]\[/b\]  
  
\[b\]\[size=large\]Java一直以来，对多线程的开发支持比较良好，特别在JDK5,6后引入的java.util.concurrent包，使用多线程的开发变的更加容易，这个包里面大部分的API都是更进一步的封装，作为开发者，我们只需熟悉它怎么使用，就能够很轻松的上手，当然如果你想完全搞懂它，那么就需要读读那些优秀的源码了。\[/size\]\[/b\]  
  
\[b\]\[size=large\]ForkJoinPool这个类是JDK7后新增的线程池，很适合在单机多核的PC上部署多线程程序，ForkJoinPool使用的分而治之的思想，这一点与当前很火的大数据处理框架Hadoop的map/reduce思想非常类似，但是他们的使用场合却不一样，ForkJoinPool适合在一台PC多核CPU上运行，而hadoop则适合在分布式环境中进行大规模集群部署。\[/size\]\[/b\]  
  
\[b\]\[size=large\]Fork/Join 模式有自己的适用范围。如果一个应用能被分解成多个子任务，并且组合多个子任务的结果就能够获得最终的答案，那么这个应用就适合用 Fork/Join 模式来解决.\[/size\]\[/b\]  
  
\[b\]\[size=large\]ForkJoinPool使用的工作窃取的方式能够在最大方式上充分利用CPU的资源，一般流程是fork分解，join结合。本质是将一个任务分解成多个子任务，每个子任务用单独的线程去处理，主要几个常用方法有\[/size\]\[/b\]  
\[table\]  
|fork( ForkJoinTask)|异步执行一个线程  
|join( ForkJoinTask)|等待任务完成并返回执行结果  
|execute( ForkJoinTask)|执行不带返回值的任务  
|submit( ForkJoinTask)|执行带返回值的任务  
|invoke( ForkJoinTask)|执行指定的任务，等待完成，返回结果。  
|invokeAll(ForkJoinTask)|执行指定的任务，等待完成，返回结果。  
|shutdown()|执行此方法之后，ForkJoinPool 不再接受新的任务，但是已经提交的任务可以继续执行。如果希望立刻停止所有的任务，可以尝试 shutdownNow() 方法。  
|awaitTermination(int, TimeUnit.SECONDS)|阻塞当前线程直到 ForkJoinPool 中所有的任务都执行结束。  
|compute（）|执行任务的具体方法  
| Runtime.getRuntime().availableProcessors()|获取CPU个数的方法  
\[/table\]  
\[b\]\[size=x-large\]关于上表中的ForkJoinTask是一个抽象类，代表一个可以fork与join的任务. ,它还有2个抽象子类:RecursiveAcion和RecursiveTask.其中RecursiveTask代表有泛型返回值的任务.而RecursiveAction代表没有返回值.  
\[/size\]\[/b\]  
  
\[b\]\[size=large\]下面给出测试代码\[/size\]\[/b\]

package com.demo;
    
    import java.util.concurrent.RecursiveAction;
    
    /**
     * 
     * 继承RecursiveAction来实现可分解的任务
     * 注意无返回值
     * 
     * **/
    
    public class PrintTask extends RecursiveAction {
       
    
    	//每个小任务，最多只打印50个数
    	private static final int threshold=50;
    	//打印任务的开始
    	private int start;
    	//打印任务的结束
    	private int end;
    
    	public PrintTask() {
       
    		// TODO Auto-generated constructor stub
    	}
    
    
    
    	//打印从start到end之间的任务
    	public PrintTask(int start, int end) {
       
    		super();
    		this.start = start;
    		this.end = end;
    	}
    
    
    	@Override
    	protected void compute() {
       
    
    		if(end-start<threshold){
       
    			for(int i=start;i<end;i++){
       
    
    				System.out.println(Thread.currentThread().getName()+"i的值:"+i);
    			}
    		}else{
       
    			//当end与start之间的差大于threshold，及打印的数超过50个时，
    			//将大任务分解成2个小任务
    			int middle=(start+end)/2;
    			PrintTask left=new PrintTask(start, middle);
    			PrintTask  right=new PrintTask(middle, end);
    			//并行执行两个小任务
    			left.fork();
    			right.fork();
    
    		}
    
    
    	}
    
    
    
    }

package com.demo;
    
    import java.util.concurrent.ForkJoinPool;
    import java.util.concurrent.ForkJoinTask;
    import java.util.concurrent.TimeUnit;
    /**
     * 测试打印
     * 
     * */
    public class Test {
       
    
    
    	public static void main(String[] args)throws Exception {
       
    		ForkJoinPool pool=new ForkJoinPool();
    		//提交可分解的任务
    		pool.submit(new PrintTask(0,101));
    
    		//阻塞等待所有任务完成 
    		pool.awaitTermination(2, TimeUnit.SECONDS);
    		pool.shutdown();//关闭线程池
    
    	}
    }

\[b\]\[size=x-large\]有返回值的demo\[/size\]\[/b\]

package com.demo;
    
    import java.util.concurrent.RecursiveTask;
    /***
     * 有返回值
     * 
     * */
    public class CalTask  extends RecursiveTask<Integer>{
       
    
    
    
    	//将每个小任务，最多只能累加20个数
    	private static final int threshold=20;
    	private int arr[];
    	private int start;//开始
    	private int end;//
    	//累加从start到end之间的数
    
    	public CalTask() {
       
    		// TODO Auto-generated constructor stub
    	}
    
    
    	//累加从start到end的数组元素
    	public CalTask(int[] arr, int start, int end) {
       
    		super();
    		this.arr = arr;
    		this.start = start;
    		this.end = end;
    	}
    
    
    	@Override
    	protected Integer compute() {
       
    		int sum=0;
    		//当end与start之间的差小于threshold，开始进行累加
    		if(end-start<threshold){
       
    			for(int i=start;i<end;i++){
       
    				sum+=arr[i];
    			}
    			return sum;
    
    		}else{
       
    
    			//当end与start之间的差大于threshold，要计算的数超过20个时，
    			//将大任务分解成两个小任务
    
    			int middle=(start+end)/2;
    			CalTask left=new CalTask(arr, start, middle);
    			CalTask right=new CalTask(arr, middle, end);
    			//并行执行2个小任务
    			left.fork();
    			right.fork();
    			//把2个小任务，累加的结果合并起来
    			return left.join()+right.join();
    		}
    
    
    	}
    
    
    }

package com.demo;
    
    import java.util.Random;
    import java.util.concurrent.ForkJoinPool;
    import java.util.concurrent.Future;
    
    public class Sum {
       
    
    
    	public static void main(String[] args)throws Exception {
       
    
    		int []arr=new int[1100];
    		Random rand=new Random();
    		int total=0;
    		//初始化100个数字
    		for(int i=0;i<arr.length;i++){
       
    			int tmp=rand.nextInt(20);
    			//对元素赋值，并将数组元素的值添加到total总和中
    			total+=(arr[i]=tmp);
    
    
    		}
    		System.out.println("正确的total:"+total);
    		ForkJoinPool pool=new ForkJoinPool();
    		//提交可分解的CalTask任务
    		Future<Integer> future=pool.submit(new CalTask(arr, 0, arr.length));
    		System.out.println(future.get());
    		//关闭线程池
    
    		pool.shutdown();
    	}
    
    }