Future 和 ExecutorCompletionService 对比和使用

Posted 2020-08-27 路漫漫，其修远，吾将上下而求索

附加：Java 4种线程池介绍请查看 

谈谈new Thread的弊端及Java四种线程池的使用

当我们通过Executor提交一组并发执行的任务，并且希望在每一个任务完成后能立即得到结果，有两种方式可以采取：

 

方式一：

通过一个list来保存一组future，然后在循环中轮训这组future,直到每个future都已完成。如果我们不希望出现因为排在前面的任务阻塞导致后面先完成的任务的结果没有及时获取的情况，那么在调用get方式时，需要将超时时间设置为0

public class CompletionServiceTest {  
  
    static class Task implements Callable<String>{  
        private int i;  
          
        public Task(int i){  
            this.i = i;  
        }  
  
        @Override  
        public String call() throws Exception {  
            Thread.sleep(10000);  
            return Thread.currentThread().getName() + "执行完任务：" + i;  
        }     
    }  
      
    public static void main(String[] args){  
        testUseFuture();  
    }  
      
    private static void testUseFuture(){  
        int numThread = 5;  
        ExecutorService executor = Executors.newFixedThreadPool(numThread);  
        List<Future<String>> futureList = new ArrayList<Future<String>>();  
        for(int i = 0;i<numThread;i++ ){  
            Future<String> future = executor.submit(new CompletionServiceTest.Task(i));  
            futureList.add(future);  
        }  
                  
        while(numThread > 0){  
            for(Future<String> future : futureList){  
                String result = null;  
                try {  
                    result = future.get(0, TimeUnit.SECONDS);  
                } catch (InterruptedException e) {  
                    e.printStackTrace();  
                } catch (ExecutionException e) {  
                    e.printStackTrace();  
                } catch (TimeoutException e) {  
                    //超时异常直接忽略 
　　　　　　　　　　　　//future.cancel(true);//超时设置任务取消
                }  
                if(null != result){  
                    futureList.remove(future);  
                    numThread--;  
                    System.out.println(result);  
                    //此处必须break，否则会抛出并发修改异常。（也可以通过将futureList声明为CopyOnWriteArrayList类型解决）  
                    break;  
                }  
            }  
        }  
    }  
}  

 

 方式二：

第一种方式显得比较繁琐，通过使用ExecutorCompletionService，则可以达到代码最简化的效果。

public class CompletionServiceTest {  
  
    static class Task implements Callable<String>{  
        private int i;  
          
        public Task(int i){  
            this.i = i;  
        }  
  
        @Override  
        public String call() throws Exception {  
            Thread.sleep(10000);  
            return Thread.currentThread().getName() + "执行完任务：" + i;  
        }     
    }  
      
    public static void main(String[] args) throws InterruptedException, ExecutionException{  
        testExecutorCompletionService();  
    }  
      
    private static void testExecutorCompletionService() throws InterruptedException, ExecutionException{  
        int numThread = 3;  
        ExecutorService executor = Executors.newFixedThreadPool(numThread);  
        CompletionService<String> completionService = new ExecutorCompletionService<String>(executor);  
        for(int i = 0;i<numThread;i++ ){  
            completionService.submit(new CompletionServiceTest.Task(i));  
        }  
}  
          
        for(int i = 0;i<numThread;i++ ){       
            System.out.println(completionService.take().get());  //获取执行结果
        }  
          
    }  

 

ExecutorCompletionService分析：

 CompletionService是Executor和BlockingQueue的结合体。

    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<Future<V>>();  
    }  

 

任务的提交和执行都是委托给Executor来完成。在构造函数中创建一个BlockingQueue来保存计算完成的结果，当提交某个任务时，该任务首先将被包装为一个QueueingFuture，

    public Future<V> submit(Callable<V> task) {  
        if (task == null) throw new NullPointerException();  
        RunnableFuture<V> f = newTaskFor(task);  
        executor.execute(new QueueingFuture(f));  
        return f;  
    }  

 QueueingFuture，这是FutureTask的一个子类，通过改写该子类的done方法，可以实现当任务完成时，将结果放入到BlockingQueue中。

  

private class QueueingFuture extends FutureTask<Void> {

        QueueingFuture(RunnableFuture<V> task) {

            super(task, null);

            this.task = task;

        }

        protected void done() { completionQueue.add(task); }

        private final Future<V> task;

    }

 

 

 而通过使用BlockingQueue的take(阻塞获取)或poll(非阻塞获取)方法，则可以得到结果。在BlockingQueue不存在元素时，这两个操作会阻塞，一旦有结果加入，则立即返回。

附加知识点：

take()：取走BlockingQueue里排在首位的对象,若BlockingQueue为空,阻断进入等待状态直到Blocking有新的对象被加入为止;

poll(time):取走BlockingQueue里排在首位的对象,若不能立即取出,则可以等time参数规定的时间,取不到时返回nul

public Future<V> take() throws InterruptedException {  
    return completionQueue.take();  
}  
  
public Future<V> poll() {  
    return completionQueue.poll();  
}