ExecutorCompletionService使用详解

ExecutorCompletionService使用详解 前言

ExecutorCompletionService 为什么需要使用这个类？

jdk8提供的 CompletionService 接口，实现类 ExecutorCompletionService。

我们先来看一个例子？
很明显下面任务批量提交线程池，等待线程池运行，每个任务执行时间不同。

        //此线程池线程递增
        ExecutorService executorService = Executors.newCachedThreadPool();

        List<Callable<Integer>> list = new ArrayList<>();
   list.add(() -> {
            TimeUnit.SECONDS.sleep(5);
            System.out.println(" 5s ");
            return 5;
        });

        list.add(() -> {
            TimeUnit.SECONDS.sleep(1);
            System.out.println(" 1s ");
            return 1;
        });
        list.add(() -> {
            TimeUnit.SECONDS.sleep(8);
            System.out.println(" 8s ");
            return 8;
        });

批量提交，批量返回

 List<Future<Integer>> invokeAll = executorService.invokeAll(list);

System.out.println(invokeAll);

乍一看感觉没什么问题，运行也没问题。
但实际上，我们根本不知道哪个任务先完成，每个任务执行的时间也不一样，这就必须等执行最长的那个任务执行结束，整体才能执行完。
最先执行结束的任务会一直等待，最晚执行的那个任务结束。

可以看到，这些执行结束后，等虽有的都结束后，结果才打印出来。

批量任务

假设每个任务执行时间
1 : 10s
2 : 2s
3 : 5s
4 : 10s
在 几个任务并行情况下,第二个2s完成,但是在获取的时候别的 其他线程完成时间是最多是10s, 第二个必须等待10s 才能返回.
如果最后一个是10s完成 其他完成都在10s内, 其他线程可以做空闲 或者做其他事情,最后一个必须等10s

如何解决这个问题？

• 默认提交时候，for循环去提交，然后等待每个任务返回。

        List<Integer> newList1  = new ArrayList<>();
        for (Callable<Integer> integerCallable : list) {
            Future<Integer> submit = executorService.submit(integerCallable);
            newList1.add(submit.get());
        }

• 使用 ExecutorCompletionService

ExecutorCompletionService 如何使用 ？

ExecutorCompletionService使用场景

解决批量提交任务办法就是使用使用 ExecutorCompletionService，异步通知返回。

当前类提供的方法

提供的两个构造函数，一个可以指定返回阻塞队列，另一个使用默认的。另外都需要提供一个线程池进来。

提供了三个获取方法，可以看到都是从队列中获取

两个提交任务方法

如何执行任务结果放入队列呢？

可以看到是将 执行结果放入队列中。
内部实现了异步执行接口，以及重写了它的done方法

实例

还是提交上面的任务

        List<Future<Integer>> futureList  = new ArrayList<>();
        CompletionService completionService  =  new ExecutorCompletionService(executorService);
        for (Callable<Integer> integerCallable : list) {
            Future submit = completionService.submit(integerCallable);
           futureList.add(submit);

        }

take 获取

        //获取 谁先执行完 谁先出来 take() 获取时候回阻塞  也可以通过Poll方法获取
        while ((completionService.take())!=null){
            System.out.println(completionService.take().get());
        }

poll 方法获取

        Future poll;
        while ((poll = completionService.poll(10, TimeUnit.SECONDS)) != null) {
            System.out.println(poll.get());
        }

poll 和 take 区别在于 poll可以执行超时时间

可以看到，谁先执行结束 谁先出来。

最后

ExecutorCompletionService 相比之前Future 相比 ，提供了一个通知机制，不会让你傻傻的在哪儿等着获取，而是将结果统一让如一个队列，当前提交任务不会阻塞获取，从另一个队列中阻塞获取。