Java：多线程，线程池，使用CompletionService通过Future来处理Callable的返回结果

你的名字 2021-05-12 11:51 427阅读 0赞

## 1. 背景 ##

在Java5的多线程中，可以使用Callable接口来实现具有返回值的线程。使用线程池的submit方法提交Callable任务，利用submit方法返回的Future存根，调用此存根的get方法来获取整个线程池中所有任务的运行结果。

方法一：如果是自己写代码，应该是自己维护一个Collection保存submit方法返回的Future存根，然后在主线程中遍历这个Collection并调用Future存根的get()方法取到线程的返回值。

方法二：使用CompletionService类，它整合了Executor和BlockingQueue的功能。你可以将Callable任务提交给它去执行，然后使用类似于队列中的take方法获取线程的返回值。

## 2. 实现代码 ##

![复制代码][copycode.gif]

package com.clzhang.sample.thread;
    
    import java.util.*;
    import java.util.concurrent.BlockingQueue;
    import java.util.concurrent.Callable;
    import java.util.concurrent.CompletionService;
    import java.util.concurrent.ExecutorCompletionService;
    import java.util.concurrent.ExecutorService;
    import java.util.concurrent.Executors;
    import java.util.concurrent.Future;
    import java.util.concurrent.LinkedBlockingQueue;
    
    public class ThreadPoolTest4 {
        // 具有返回值的测试线程
        class MyThread implements Callable<String> {
            private String name;
            public MyThread(String name) {
                this.name = name;
            }
    
            @Override
            public String call() {
                int sleepTime = new Random().nextInt(1000);
                try {
                    Thread.sleep(sleepTime);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
    
                // 返回给调用者的值
                String str = name + " sleep time:" + sleepTime;
                System.out.println(name + " finished...");
    
                return str;
            }
        }
    
        private final int POOL_SIZE = 5;
        private final int TOTAL_TASK = 20;
    
        // 方法一，自己写集合来实现获取线程池中任务的返回结果
        public void testByQueue() throws Exception {
            // 创建线程池
            ExecutorService pool = Executors.newFixedThreadPool(POOL_SIZE);
            BlockingQueue<Future<String>> queue = new LinkedBlockingQueue<Future<String>>();
    
            // 向里面扔任务
            for (int i = 0; i < TOTAL_TASK; i++) {
                Future<String> future = pool.submit(new MyThread("Thread" + i));
                queue.add(future);
            }
    
            // 检查线程池任务执行结果
            for (int i = 0; i < TOTAL_TASK; i++) {
                System.out.println("method1:" + queue.take().get());
            }
    
            // 关闭线程池
            pool.shutdown();
        }
    
        // 方法二，通过CompletionService来实现获取线程池中任务的返回结果
        public void testByCompetion() throws Exception {
            // 创建线程池
            ExecutorService pool = Executors.newFixedThreadPool(POOL_SIZE);
            CompletionService<String> cService = new ExecutorCompletionService<String>(pool);
    
            // 向里面扔任务
            for (int i = 0; i < TOTAL_TASK; i++) {
                cService.submit(new MyThread("Thread" + i));
            }
    
            // 检查线程池任务执行结果
            for (int i = 0; i < TOTAL_TASK; i++) {
                Future<String> future = cService.take();
                System.out.println("method2:" + future.get());
            }
    
            // 关闭线程池
            pool.shutdown();
        }
    
        public static void main(String[] args) throws Exception {
            ThreadPoolTest4 t = new ThreadPoolTest4();
            t.testByQueue();
            t.testByCompetion();
        }
    }

![复制代码][copycode.gif]

部分输出：

...

Thread4 finished...  
method1:Thread4 sleep time:833  
method1:Thread5 sleep time:158  
Thread6 finished...  
method1:Thread6 sleep time:826  
method1:Thread7 sleep time:185  
Thread9 finished...  
Thread8 finished...  
method1:Thread8 sleep time:929  
method1:Thread9 sleep time:575

...

Thread11 finished...  
method2:Thread11 sleep time:952  
Thread18 finished...  
method2:Thread18 sleep time:793  
Thread19 finished...  
method2:Thread19 sleep time:763  
Thread16 finished...  
method2:Thread16 sleep time:990

...

## 3. 总结 ##

使用方法一，自己创建一个集合来保存Future存根并循环调用其返回结果的时候，主线程并不能保证首先获得的是最先完成任务的线程返回值。它只是按加入线程池的顺序返回。因为take方法是阻塞方法，后面的任务完成了，前面的任务却没有完成，主程序就那样等待在那儿，只到前面的完成了，它才知道原来后面的也完成了。

使用方法二，使用CompletionService来维护处理线程不的返回结果时，主线程总是能够拿到最先完成的任务的返回值，而不管它们加入线程池的顺序。

[copycode.gif]: /images/20210511/4f90c9898d064542b9b95593e7b70be4.png