CompletionService实践

在之前的ExecutorCompletionService中，我们了解了其实现的原理。通过FutureTask对提交任务的封装和代理然后通过call方法进行回调并将返回值存储在FutureTask中的Object中，当获取值的时候采用阻塞队列方式进行存储。其中的阻塞队列是对FutureTask的引用，也就是说提交的任务是有序的。获取到的值也是有顺序的。

在工作中其实在某些业务场景下需要使用ComletionService进行并行处理。因为其返回值得有序性但是对外调用的并行性相对于单线程模式的for循环，其优点不言而喻。借此机会在此做一个小demo，以供日后工作借鉴。这里要注意的是CompletionService是一个接口，其背后的大Boss是ExecutorCompletionService。

public class RealDoSomeThing implements Callable<String> {

    private String tool;
    /***
     * 您可以通过构造方法，在任务初始化的时候传入最后调用call方法的所需要的工具类或者spring类
     */
    public RealDoSomeThing(String tool){
        this.tool=tool;
    }

    /***
     * 这里实现我们的业务，上边指定的String类型，表示返回值得类型
     * @return
     * @throws Exception
     */
    @Override
    public String call() throws Exception {
        return this.tool+"---"+"真正的执行方法";
    }
}

public class ComplationServiceTest {

    public static void main(String[] args) throws InterruptedException, ExecutionException {
        ExecutorService executorService= Executors.newFixedThreadPool(3);
        CompletionService<String> completionService=new ExecutorCompletionService(executorService);
        for (int i=0;i<10;i++){
            completionService.submit(new RealDoSomeThing("工具"+i));
        }

        for (int i=0;i<10;i++){
            String result=completionService.take().get();
            System.out.println(result);
        }
        executorService.shutdown();
    }
}

但是通过实践，我们发现所谓的返回结果的顺序性并没有体现出来。那么问题出到哪里了？通过之前的分析确实是把task任务进行缓存到了CompletionQueue队列中了。而获取值的时候就是走的ComletionQueue队列啊，可是实际并不是这样的。why？

事出反常必有妖，咋再看看源码。

咋看看咱们的宝贝ExecutorCompletionService是怎么操作的。

这下明白了吧，这玩意是在返回结果之后才加入的completionQueue队列。那么我们如何让它保持有序？如果按照我们当时的想法来说，它本就是有序的，但是它给安排到了最后才入队列。那么咋就让他在创建在没有变成线程之前就入队列。也就是我们自定义队列，然后submit之后就加入到队列中，因为java是引用。所以只要记录了引用，你跑的再远也能找得到。而submit方法本身是有返回值的。那咋就直接将返回值入队列就OK了。就是这么完美。

改造之后的具有顺序的CompletionService：

public class ComplationServiceTest {
    public static void main(String[] args) throws InterruptedException, ExecutionException {
        LinkedBlockingQueue<java.util.concurrent.Future> blockingDeque= new LinkedBlockingQueue<>();
        ExecutorService executorService= Executors.newFixedThreadPool(3);
        CompletionService<String> completionService=new ExecutorCompletionService(executorService);
        for (int i=0;i<10;i++){
            java.util.concurrent.Future future=completionService.submit(new RealDoSomeThing("工具"+i));
            blockingDeque.add(future);
        }

        for (int i=0;i<10;i++){
            String result= (String) blockingDeque.take().get();
            System.out.println(result);
        }
        executorService.shutdown();
    }
}

通过自定义的任务队列缓存，然后从自定义任务队列中获取返回值，绕过ComletionService提供的任务队列就实现了上述需求。在此真的对阅读源码感到一份实实在在的力量感。所以还是要阅读源码，知其然还要知其所以然。加油