Java8的CompletionService使用与原理

为了防止无良网站的爬虫抓取文章，特此标识，转载请注明文章出处。LaplaceDemon/ShiJiaqi。

https://www.cnblogs.com/shijiaqi1066/p/10454237.html

CompletionService是Java8的新增接口，JDK为其提供了一个实现类ExecutorCompletionService。这个类是为线程池中Task的执行结果服务的，即为Executor中Task返回Future而服务的。CompletionService的实现目标是任务先完成可优先获取到，即结果按照完成先后顺序排序。

CompletionService的使用非常简单。从源码查看ExecutorCompletionService类，该类只有三个成员变量：

public class ExecutorCompletionService<V> implements CompletionService<V> {
    private final Executor executor;
    private final AbstractExecutorService aes;
    private final BlockingQueue<Future<V>> completionQueue;
    
    ...
}

可以看到ExecutorCompletionService主要是增强executor线程池的。Task包装后被塞入completionQueue，当Task结束，其Future就可以从completionQueue中获取到。

其基本原理可以参看下图：

CompletionService接口源码：

public interface CompletionService<V> {
    // 提交
    Future<V> submit(Callable<V> task);
    Future<V> submit(Runnable task, V result);
    // 获取
    Future<V> take() throws InterruptedException;
    Future<V> poll();
    Future<V> poll(long timeout, TimeUnit unit) throws InterruptedException;
}

例：向CompletionService中提交10个Task，当Task有任务返回则会优先从CompletionService内部的队列中获取到Task的Future。

package test;

import java.util.ArrayList;
import java.util.Date;
import java.util.List;
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;

public class TestCompletionService {
	public static void main(String[] args)  {
        Long start = System.currentTimeMillis();
        //开启3个线程
        ExecutorService exs = Executors.newFixedThreadPool(5);
        try {
            int taskCount = 10;
            // 结果集
            List<Integer> list = new ArrayList<Integer>();
            List<Future<Integer>> futureList = new ArrayList<Future<Integer>>();
            
            // 1.定义CompletionService
            CompletionService<Integer> completionService = new ExecutorCompletionService<Integer>(exs);  
            
            // 2.添加任务
            for(int i=0;i<taskCount;i++){
            	Future<Integer> future = completionService.submit(new Task(i+1));
                futureList.add(future);
            }

            // 3.获取结果
            for(int i=0;i<taskCount;i++){
                Integer result = completionService.take().get();
                System.out.println("任务i=="+result+"完成!"+new Date());
                list.add(result);
            }
            
            System.out.println("list="+list);
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            //关闭线程池
            exs.shutdown();
        }
        
    }

    static class Task implements Callable<Integer>{
        Integer i;
        
        public Task(Integer i) {
            super();
            this.i=i;
        }

        @Override
        public Integer call() throws Exception {
            if(i==5) {
                Thread.sleep(5000);
            }else{
                Thread.sleep(1000);
            }
            System.out.println("线程："+Thread.currentThread().getName()+"任务i="+i+",执行完成！");  
            return i;
        }
        
    }
}