Java7之线程池ForkJoinPool

许多情况下，在一个程序中使用多线程是有益处的，可以大大提高程序的效率，多线程主要有以下3个优点1,资源利用率更好2，程序设计在某些情况下更简单3，程序响应更快。当然凡事有利就有弊，多线程也会使程序的开发，维护及调试更加复杂，当然如果我们能够扬长避短，在正确的场合下使用多线程，那么它将成为我们程序中开发的利器。


Java一直以来，对多线程的开发支持比较良好，特别在JDK5,6后引入的java.util.concurrent包，使用多线程的开发变的更加容易，这个包里面大部分的API都是更进一步的封装，作为开发者，我们只需熟悉它怎么使用，就能够很轻松的上手，当然如果你想完全搞懂它，那么就需要读读那些优秀的源码了。

ForkJoinPool这个类是JDK7后新增的线程池，很适合在单机多核的PC上部署多线程程序，ForkJoinPool使用的分而治之的思想，这一点与当前很火的大数据处理框架Hadoop的map/reduce思想非常类似，但是他们的使用场合却不一样，ForkJoinPool适合在一台PC多核CPU上运行，而hadoop则适合在分布式环境中进行大规模集群部署。

Fork/Join 模式有自己的适用范围。如果一个应用能被分解成多个子任务，并且组合多个子任务的结果就能够获得最终的答案，那么这个应用就适合用 Fork/Join 模式来解决.

ForkJoinPool使用的工作窃取的方式能够在最大方式上充分利用CPU的资源，一般流程是fork分解，join结合。本质是将一个任务分解成多个子任务，每个子任务用单独的线程去处理，主要几个常用方法有

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个数的方法

关于上表中的ForkJoinTask是一个抽象类，代表一个可以fork与join的任务. ,它还有2个抽象子类:RecursiveAcion和RecursiveTask.其中RecursiveTask代表有泛型返回值的任务.而RecursiveAction代表没有返回值.


下面给出测试代码

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();//关闭线程池
		
	}
}

有返回值的demo

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();
	}

}

评论

1 楼 ksisn 2014-11-22  

楼主您好，很荣幸能拜读到您的博客，我是刚毕业的程序猿，用您这篇文章的例子进行一些测试，发现比单线程方式在时间上慢了很多倍，那么像这种ForkJoinPool 适用的场景又是什么呢，一直在拜读您的例子中，这个没弄懂，特地请教您。