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 completionService; private int cpuCoreNumber; public MyCompletionService() {// 获取当前系统的CPU 数目 cpuCoreNumber = Runtime.getRuntime().availableProcessors();// ExecutorService通常根据CPU数目定义线程池大小 executorService = Executors.newFixedThreadPool(cpuCoreNumber);// 更高级的ExecutorService completionService = new ExecutorCompletionService (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 { 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(); }}
运行结果：
