java中多线程的使用-Fork_Join - 没昵称可用

Fork-Join的介绍

　　Fork/Join框架是Java7提供了的一个用于并行执行任务的框架， 是一个把大任务分割成若干个小任务，最终汇总每个小任务结果后得到大任务结果的框架。

我们再通过Fork和Join这两个单词来理解下Fork/Join框架，Fork就是把一个大任务切分为若干子任务并行的执行，Join就是合并这些子任务的执行结果，最后得到这个大任务的结果。比如计算1+2+。。＋10000，可以分割成10个子任务，每个子任务分别对1000个数进行求和，最终汇总这10个子任务的结果。

工作窃取算法

　　假如我们需要做一个比较大的任务，我们可以把这个任务分割为若干互不依赖的子任务，为了减少线程间的竞争，于是把这些子任务分别放到不同的队列里，并为每个队列创建一个单独的线程来执行队列里的任务，线程和队列一一对应，比如A线程负责处理A队列里的任务。但是有的线程会先把自己队列里的任务干完，而其他线程对应的队列里还有任务等待处理。干完活的线程与其等着，不如去帮其他线程干活，于是它就去其他线程的队列里窃取一个任务来执行。而在这时它们会访问同一个队列，所以为了减少窃取任务线程和被窃取任务线程之间的竞争，通常会使用双端队列，被窃取任务线程永远从双端队列的头部拿任务执行，而窃取任务。

Fork/Join框架

• ForkJoinTask：我们要使用ForkJoin框架，必须首先创建一个ForkJoin任务。它提供在任务中执行fork()和join()操作的机制，通常情况下我们不需要直接继承ForkJoinTask类，而只需要继承它的子类，Fork/Join框架提供了以下两个子类：
  • RecursiveAction：用于没有返回结果的任务。
  • RecursiveTask ：用于有返回结果的任务。
• ForkJoinPool ：ForkJoinTask需要通过ForkJoinPool来执行，任务分割出的子任务会添加到当前工作线程所维护的双端队列中，进入队列的头部。当一个工作线程的队列里暂时没有任务时，它会随机从其他工作线程的队列的尾部获取一个任务。  

Fork_Join实例

package forkjoin;

import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.ForkJoinTask;
import java.util.concurrent.RecursiveTask;

public class ForkJoinService{

	private static ForkJoinService instance = null;
	private ForkJoinPool forkjoinpool;//线程池
	private int maxPoolSize;//最大线程数
	public int getMaxPoolSize() {
		return maxPoolSize;
	}
	public void setMaxPoolSize(int maxPoolSize) {
		this.maxPoolSize = maxPoolSize;
	}
	public ForkJoinService(){//获取本系统的有效线程数，设置线程池为有效线程的两倍。
		int size = Runtime.getRuntime().availableProcessors();
		maxPoolSize = size*2;
		forkjoinpool = new ForkJoinPool(maxPoolSize);
		instance = this;
	}
	private void stop(){
		if(forkjoinpool != null)
			forkjoinpool.shutdown();
	}
	public <T> ForkJoinTask<T> submit(ForkJoinTask<T> task){//将任务提交给线程，去执行。
		return forkjoinpool.submit(task);
	}
	public <T> T invoke(ForkJoinTask<T> task){
		return forkjoinpool.invoke(task);
	}
}
package forkjoin;

import java.util.Collections;
import java.util.List;
import java.util.concurrent.RecursiveTask;

import org.springframework.beans.factory.annotation.Autowired;

import entity.Teacher;
import service.TeacherService;

public class PersonTask extends RecursiveTask<List<Teacher>> {

	@Autowired
	private TeacherService teacherService;
	private String[] ids;
	public PersonTask(String[] ids){
		this.ids = ids;
	}
	@Override
	protected List<Teacher> compute() {
		// TODO Auto-generated method stub
		List<Teacher> rtn = null;
		if(ids == null)
			return Collections.emptyList();
		for(int i=0;i<3;i++){
			try {
				rtn = teacherService.findByids(ids);
			} catch (Exception e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
			if(rtn!=null){
				break;
			}
		}
		return rtn;
	}

}
package forkjoin;

import java.util.List;

import entity.Teacher;
import net.sf.json.JSONObject;

public class ForkJoinTest {

	public static void main(String[] args){
		ForkJoinService forkJoinService = new ForkJoinService();
		List<String> ids = null;
		int pagesize = 5;
		int size = ids.size();
		for(int i=0;i<size;i+=pagesize){
			int end = i+pagesize;
			if(end>size) end=size;
			String[] array = ids.subList(i, end).toArray(new String[end-i]);
			List<Teacher> list = (List<Teacher>) forkJoinService.submit(new PersonTask(array));
			JSONObject join = new JSONObject();
			join.put("teacher", list);
			System.out.println("youzhongmin========="+join.toString());
		}
	}
}