多线程在需要返回值时，我们知道需要用到Callable和Future。Callable的cell方法可以返回一个值并且可抛出异常，是对Runnable的很好的补充；Future表示了一个任务的周期，它提供了判断任务状态、获取任务结果和取消任务等方法 。
下面演示三种使用Executor执行Callable任务的方法。/** * 测试任务，返回任务的序号 */ public static class TestTask implements Callable{ int index; public TestTask(int index) { this.index = index; } @Override public Integer call() throws Exception { return index; } } /** * 方法一：手动的保存任务的返回，这样的好处是每个任务对应的结果我们很清楚 */ @Test public void ordinaryTest() throws ExecutionException, InterruptedException { ExecutorService es = Executors.newFixedThreadPool(10); List > futures = new ArrayList<>(); for(int i = 0; i < 10; i++) { TestTask testTask = new TestTask(i); Future future = es.submit(testTask); futures.add(future); } es.shutdown(); for(int i = 0; i < 10; i++) { System.out.println("index:" + i + ",future:"+ futures.get(i).get()); } } /** * 方法二：使用ExecutorCompletionService * ExecutorCompletionService中使用阻塞队列保存各任务的返回结果，返回是无序的，即谁先执行完成（异常、中断），谁先入队。 * 当我们不关心结果的顺序，或者需要一个任务完成时就取消其他任务的情况下，它是非常的方便的 */ @Test public void completionServiceTest() throws ExecutionException, InterruptedException { ExecutorService es = Executors.newFixedThreadPool(10); CompletionService completionService = new ExecutorCompletionService<>(es); for(int i = 0; i < 10; i++) { TestTask testTask = new TestTask(i); completionService.submit(testTask); } es.shutdown(); for(int i = 0; i < 10; i++) { Future future = completionService.take(); System.out.println("index:" + i + ",future:"+ future.get()); } } /** * 方法三：ExecutorService的invokeAll方法 * invokeAll方法入参为一组任务，返回一组Future，这两个集合是有相同结构的， * 即它是按照入参的任务集合中迭代器的顺序将所有的Future添加到返回的集合中，从而任务和Future在它们各自的集合中有着同样的顺序。 * 当我们需要任务和结果的对应关系时，使用invokeAll方法来代替第一种方法 */ @Test public void invokeAllTest() throws InterruptedException, ExecutionException { ExecutorService es = Executors.newFixedThreadPool(10); List tasks = new ArrayList<>(); for (int i = 0; i < 10; i++){ tasks.add(new TestTask(i)); } List > futures = es.invokeAll(tasks); es.shutdown(); for (int i = 0; i < futures.size(); i++){ System.out.println("index:" + i + ",future:"+ futures.get(i).get()); } }
下面看一下ExecutorCompletionService的原理：
ExecutorCompletionService是将Executor和的功能融合在一起，可将Callbale任务提交给它来执行，然后我们就可以像队列一样使用take或poll来得到已经完成的任务结果。下面是源码分析：/** *ExecutorCompletionService包含三个成员变量，最主要的是completionQueue，它的类型阻塞队列 */ private final Executor executor; private final AbstractExecutorService aes; private final BlockingQueue> completionQueue; /** * 构造方法，需要我们传入一个Executor */ public ExecutorCompletionService(Executor executor) { if (executor == null) throw new NullPointerException(); this.executor = executor; this.aes = (executor instanceof AbstractExecutorService) ? (AbstractExecutorService) executor : null; this.completionQueue = new LinkedBlockingQueue >(); } /** * 提交任务的方法，其中的RunnableFuture为一个内部类，继承自FutureTask */ public Future submit(Callable task) { if (task == null) throw new NullPointerException(); RunnableFuture f = newTaskFor(task); executor.execute(new QueueingFuture(f)); return f; } private class QueueingFuture extends FutureTask { QueueingFuture(RunnableFuture task) { super(task, null); this.task = task; } /** * 这是QueueingFuture存在的主要原因，当任务执行完成后，将任务结果装入队列中 */ protected void done() { completionQueue.add(task); } private final Future task; } /** * 从队列中获取返回值 */ public Future take() throws InterruptedException { return completionQueue.take(); }