java多线程Executor

在处理大批量数据的时候，我习惯性使用Executor,将一批数据按拆分到多个线程上，每个线程保证数据隔离,每个单元都是相互独立的，使用场景：工单处理、用户额度计算。
下面，我以求一批数据最大值为例写一个多线程处理.

package com.yxkong.demo.executor;
import java.util.ArrayList;
import java.util.List;
import java.util.Random;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.FutureTask;


public class ExecutorThread {
	/**
	 * 线程池
	 */
	private ExecutorService exec;
	/**
	 * cpu 个数
	 */
	private int cpuNum;
	/**
	 * 接收线程处理的返回值
	 */
	private List<Future<Integer>> tasks = new ArrayList<Future<Integer>>();
	
	public ExecutorThread(ExecutorService exec, int cupNum) {
		this.cpuNum = cupNum;
		this.exec = exec;
	}
	/**
	 * 有返回值的
	 * @author ducongcong
	 * @date 2017年5月21日
	 */
	class SumExecutor implements Callable<Integer> {
		
		//拆分后的任务
		private List<Integer> list;
		public SumExecutor(List<Integer> list) {
			this.list = list;
		}
		public Integer call() throws Exception {
			String threadName = Thread.currentThread().getName();
			Integer maxId = 0;
			//处理当前线程上的任务,list数据可以是一个个独立事务的方法，可以是单个线程的方法
			for(Integer i:list){
				if(maxId<i){
					maxId = i;
				}
			}
			System.err.println(threadName+" maxId="+maxId);
			return maxId;
		}
		
	}

	/**
	 * 给cpu的每个核心分配任务
	 */
	public Integer exeData(List<Integer> dataTasks) {
		// 根据CPU核心个数拆分任务，创建FutureTask并提交到Executor
		SumExecutor subCalc = null;
		FutureTask<Integer> task = null;
		int size = dataTasks.size();
		for (int i = 0; i < cpuNum; i++) {
			int increment =  size/ cpuNum + 1;
			int start = increment * i;
			int end = increment * i + increment;
			if (end > size)
				end = size;
			//拆分子任务，并将子任务提交给线程池
			subCalc = new SumExecutor(dataTasks.subList(start, end));
			task = new FutureTask<Integer>(subCalc);
			tasks.add(task);
			exec.submit(task);
		}
		subCalc = null;
		task = null;
		return getMaxId();
	}
	/**
	 * 获取最大值
	 */
	public Integer getMaxId() {
		Integer result = 0;
		for (Future<Integer> task : tasks) {
			try {
				if (result < task.get()) {
					result = task.get();
				}
			} catch (Exception e) {
				e.printStackTrace();
			}
		}
		return result;
	}
	private static List<Integer> genList(int size) {
		List<Integer> list = new ArrayList<Integer>();
        for (int i = 0; i <= size; i++) {
        	list.add(new Random().nextInt());
        }
        return list;
	}
	public static void main(String[] args) {
		//生产上executorService是一个公用的
		int cpuNum = 4;
		ExecutorService executorService = Executors.newFixedThreadPool(cpuNum);
		List<Integer> arrays = genList(100);
		System.err.println(arrays.toString());
		ExecutorThread executorThread = new ExecutorThread(executorService, 4);
		executorThread.exeData(arrays);
		System.err.println("max:"+ executorThread.getMaxId());
		
	}
}