CompletionService实例-Mine

package com.bugyun.test;

import java.util.concurrent.Callable;
import java.util.concurrent.CompletionService;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorCompletionService;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class MyCompletionService {
	
	private ExecutorService executorService;
	private CompletionService<Long> completionService;
	private int cpuCoreNumber;

	public MyCompletionService() {
//		获取当前系统的CPU 数目
		cpuCoreNumber = Runtime.getRuntime().availableProcessors();
//		ExecutorService通常根据CPU数目定义线程池大小
		executorService = Executors.newFixedThreadPool(cpuCoreNumber);
//		更高级的ExecutorService
		completionService = new ExecutorCompletionService<Long>(executorService);
	}
	
	
	/**
	 * 将任务分配给子任务执行
	 */
	public Long statistic(final int[] numbers) {
		int size = numbers.length;
		for (int i = 0; i < cpuCoreNumber; i++) {
			
			int increment = size / cpuCoreNumber + 1;
			int start = increment * i;
			int end = increment * (i + 1);
			if (end > size){
				end = size;
			}
			Task task = new Task(numbers, start, end);
			if(!executorService.isShutdown()){
				completionService.submit(task);
			}
		}
		return getResult();
	}

	/**
	 * 遍历子任务，获取结果
	 */
	public Long getResult() {
		Long result = 0l;
		for (int i = 0; i < cpuCoreNumber; i++) {
			try {
				Long subSum = completionService.take().get();
				System.out.println(" ===> 获取结果的核是："+i+" , 此时结果为："+subSum);
				result += subSum;
			} catch (InterruptedException e) {
				e.printStackTrace();
			} catch (ExecutionException e) {
				e.printStackTrace();
			}
		}
		return result;
	}

	public void close() {
		executorService.shutdown();
	}

	
	class Task implements Callable<Long> {
		private int[] numbers;
		private int start;
		private int end;

		public Task(final int[] numbers, int start, int end) {
			this.numbers = numbers;
			this.start = start;
			this.end = end;
		}

		public Long call() throws Exception {
			Long sum = 0l;
			for (int i = start; i < end; i++) {
				sum += numbers[i];
			}
			return sum;
		}
	}
	
	
	public static void main(String... args) throws Exception {
		int[] numbers = new int[] { 1, 2, 3, 4, 5, 6, 7, 8, 9 };
		MyCompletionService myCompletion = new MyCompletionService();
		Long result = myCompletion.statistic(numbers);
		System.out.println("最终结果是："+result);
		myCompletion.close();
	}

}

 

 运行结果：