乐趣区

关于java:呵呵面试官问我知不知道CompletionService

jiezi

这是 why 的第 61 篇原创文章

荒腔走板

大家好，我是 why，欢送来到我间断周更优质原创文章的第 61 篇。

老规矩，先荒腔走板聊聊其余的。

下面这图是上周六，成都叁缺壹演唱会的现场。正在唱歌的这个男人叫做李志。去年他因为某些起因被 404 了。上周六他忽然呈现在成都，忽然宣告着解除封印。

我晓得这个音讯是因为周六的早晨我正在看《乐队的夏天》。刚好是达达乐队正在唱《北方》，而后下面有一个弹幕：李志在成都复出啦！

于是我关上了微博，搜寻：李志成都。果然看到了他的上演视频。视频外面唱的是《对于郑州我晓得的不多》。

只是李志把歌词外面的郑州都改成了“成都”。一边是达达乐队唱着《北方》，一边是李志的《对于“成都”我晓得的不多》。那一刻，我把这个音讯分享给坐在旁边的女朋友，我张了张口，居然忽然一下有点说不出话来，只好把手机在她背后晃了一晃。

记得我之前在网抑云听李志的《梵高学生》的时候，无意间看了一下上面的评论，有一个评论的 ID 叫做放羊的壮年，他说：

往年五月十六日被确诊肺癌早期，医生说还有三到五个月，然而到明天我还呼吸着这世界的空气，人生大起大落往往都是本人领会，这算是一种孤单吗？兴许有一天我女儿也会爱上这首歌，那是不是比我更加孤单？怎么去想…

他的留言日期是 2015 年 12 月 30 日。

我点进他的主页查看，他的头像是一个很可恶的小女孩，扎着两个小辫子。可能就是他的女儿吧，我看了他的听歌排行，点了最近一周，显示没有听歌排行数据。看了他的仅有的一条动静，外面有 4000 多个评论，

热评第一是这样说的：多少人和我一样看到梵高学生下的评论而后点进来点进最近一周听歌为空白心头一凉，不晓得你当初还好不好，衷心祝愿你的女儿幸福。

他的评论外面充斥了陌生人，语言之间充斥了关心和爱，有人和他像敌人一样留言聊天，有人说他换了播放器，有人说心愿你是骗咱们的 …

评论看着看着就感觉很和煦。

隔了几天之后我也去留言了：咱们生来就是孤单，然而世间和煦。

好了，说回文章。

再谈 Future

上周不是写了《笑了，面试官问我知不知道异步编程的 Future》这一篇文章聊 Future 嘛。

而后有读者就给我留言了：why 哥，你都写到 Future 了，应该再写一下 CompletionService 的。上次面试就被面试官诘问了。

我笑着说：哎呀，切实是没有工夫写了。文章曾经很长了，再把这个货色补充下来，更长了，没人看的。

读者说：没事，我就是一个小倡议。

好的，承受倡议。本文就来聊聊 CompletionService 这个货色。

在聊它之前，咱们先回顾一下 Future 的用法。

我先问问你，当你往线程池外面提交了一组计算工作当前，你想要取得返回值。

你应该用 Executor 的什么提交办法？这个提交办法的什么重载类型？

什么？你答不上来？呸，你个渣男，上周白嫖完了就忘了？

上周的文章外面写了啊：

用 submit 的工作类型为 Callable 的或者工作类型为 Runable，还能够再传一个返回值的：

因为是一组计算工作，你想拿到返回值去搞事件。这个返回值就被封装在 Future 外面。

怎么获取呢？

调用 Future 的 get 办法，有不带超时工夫的有限期待的舔狗类型的 get，也有带超时工夫、到点就放弃的渣男类型的 get：

来一起看个例子吧：

public class JDKThreadPoolExecutorTest {public static void main(String[] args) throws Exception {ExecutorService executorService = Executors.newCachedThreadPool();  
        ArrayList<Future<Integer>> list = new ArrayList<>();  
        Future<Integer> future_15 = executorService.submit(() -> {TimeUnit.SECONDS.sleep(15);  
            System.out.println("执行时长为 15s 的执行实现。");  
            return 15;  
        });  
        list.add(future_15);  
          
        Future<Integer> future_5 = executorService.submit(() -> {TimeUnit.SECONDS.sleep(5);  
            System.out.println("执行时长为 5s 的执行实现。");  
            return 5;  
        });  
        list.add(future_5);  
          
        Future<Integer> future_10 = executorService.submit(() -> {TimeUnit.SECONDS.sleep(10);  
            System.out.println("执行时长为 10s 的执行实现。");  
            return 10;  
        });  
        list.add(future_10);  
          
        System.out.println("开始筹备获取后果");  
        for (Future<Integer> future : list) {System.out.println("future.get() =" + future.get());  
        }  
        Thread.currentThread().join();  
    }  
}

当初有三个工作，执行工夫别离是 15s/10s/5s。通过 JDK 线程池的 submit 办法提交了这三个 Callable 类型的工作。

你先眼神编译一下，心里输入一下，你想这个代码的输入后果是什么。

首先主线程把三个工作提交到线程池外面去，把对应返回的 Future 放到 List 外面存起来，而后执行“开始筹备获取后果”的输入语句。

接着进入 for 循环，在循环外面执行 future.get() 操作，阻塞期待。

看看你心里想的输入后果是不是这样的：

从这个输入后果外面，咱们能够看出问题了。很显著的木桶效应。

三个异步工作，耗时最长的最先执行，所以最先进入 list，因而当在循环中获取这个工作后果的时候 get 操作会始终阻塞，即便执行工夫为 5s/10s 的工作曾经执行实现。

好的，举个例子。设想一个场景：

假如你是一个海王，你领有泛滥一般女性朋友。你同时邀约了三位女性朋友一起吃饭。别离给她们说：你先化妆吧，好了给我说一声，我开着我的兰博基尼来接你。

小红化妆要 2 小时。小花化妆要 1 小时。小媛化妆要 30 分钟。

因为你最先给小红说的，你就始终在小红家门口等小红化妆实现。当小红化妆实现后，你接到车上，其余两位敌人早就筹备好了，在家水灵灵的等着你来接她。

这不是一个合格的海王应该有的样子。

这就是 future 在这种场景下的局限性。

依据下面的场景编码可得（代码都是间接复制粘贴就能够用的，倡议你拿进去跑一下）：

public class ExecutorCompletionServiceTest {public static void main(String[] args) throws Exception {ExecutorService executorService = Executors.newCachedThreadPool();
        ExecutorCompletionService<String> completionService =
                new ExecutorCompletionService<>(executorService);
        System.out.println("约几个妹子一起吃个饭吧。");
        completionService.submit(() -> {System.out.println("小红：好的，哥哥。我化妆要 2 个小时。等一下哦。");
            TimeUnit.SECONDS.sleep(15);
            System.out.println("小红：我 2 个小时准时化好了，哥哥来接我吧。");
            return "小红化完了。";
        });
        completionService.submit(() -> {System.out.println("小媛：好的，哥哥。我化妆要 30 分钟。等一下哦。");
            TimeUnit.SECONDS.sleep(5);
            System.out.println("小媛：我 30 分钟准时化好了，哥哥来接我吧。");
            return "小媛化完了。";
        });
        completionService.submit(() -> {System.out.println("小花：好的，哥哥。我化妆要 1 个小时。等一下哦。");
            TimeUnit.SECONDS.sleep(10);
            System.out.println("小花：我 1 个小时准时化好了，哥哥来接我吧。");
            return "小花化完了。";
        });
        TimeUnit.SECONDS.sleep(1);
        System.out.println("都告诉完, 等着吧。");
        // 循环 3 次是因为下面提交了 3 个异步工作
        for (int i = 0; i < 3; i++) {String returnStr = completionService.take().get();
            System.out.println(returnStr + "我去接她");
        }
        Thread.currentThread().join();
    }
}

输入后果如下：

说好都是一样的一般敌人的，为什么你偏偏要始终等化妆工夫最长的小红？为什么不谁动作快，就先接谁？

你看你这样操作，让小媛、小花怎么想？只能说：你是一个坏蛋了。

什么？你个海王还伪装问我“什么是海王”？

CompletionService 援救海王

还是下面的场景，当咱们引入了 CompletionService 后就显得不一样了。

先间接看用法：

`ExecutorService executorService = Executors.newCachedThreadPool();
ExecutorCompletionService<String> completionService = new ExecutorCompletionService<>(executorService);
`

用起来十分的不便，只须要用 ExecutorCompletionService 把线程池包起来。

而后提交工作的时候用 competitionService 的 submit 办法。代码如下：

public class JDKThreadPoolExecutorTest {public static void main(String[] args) throws Exception {ExecutorService executorService = Executors.newCachedThreadPool();
        ArrayList<Future<String>> list = new ArrayList<>();
        System.out.println("约几个妹子一起吃个饭吧。");
        Future<String> future_15 = executorService.submit(() -> {System.out.println("小红：好的，哥哥。我化妆要 2 个小时。等一下哦。");
            TimeUnit.SECONDS.sleep(15);
            System.out.println("小红：我 2 个小时准时化好了，哥哥来接我吧。");
            return "小红化完了。";
        });
        list.add(future_15);
        Future<String> future_5 = executorService.submit(() -> {System.out.println("小媛：好的，哥哥。我化妆要 30 分钟。等一下哦。");
            TimeUnit.SECONDS.sleep(5);
            System.out.println("小媛：我 30 分钟准时化好了，哥哥来接我吧。");
            return "小媛化完了。";
        });
        list.add(future_5);

        Future<String> future_10 = executorService.submit(() -> {System.out.println("小花：好的，哥哥。我化妆要 1 个小时。等一下哦。");
            TimeUnit.SECONDS.sleep(10);
            System.out.println("小花：我 1 个小时准时化好了，哥哥来接我吧。");
            return "小花化完了。";
        });
        list.add(future_10);
        TimeUnit.SECONDS.sleep(1);
        System.out.println("都告诉完, 等着吧。");
        for (Future<String> future : list) {System.out.println(future.get()+"我去接她。");
        }
        Thread.currentThread().join();
    }
}

你先眼神编译一下，心里输入一下 …

算了，别编译了，间接带大家看后果吧，我曾经急不可待了：

谁先化完妆，就先去接谁。

写到这里，看到这个输入后果的时候我不禁鼓起掌来。

真正的海王应该是一个工夫治理巨匠。

先比照一下输入后果：

而后比照一下两个版本代码的差别：

变动不大，甚至说微不足道。

执行 submit 办法的对象变成了 ExecutorCompletionService。

获取工作后果的办法变成了：

`String returnStr = completionService.take().get();
`

先不看原理。你就细细的品这个获取后果的办法。

completionService.take() 了个什么玩意进去，而后调用了 get 办法。

依据这个 get，直觉就通知我 take 进去的必定是一个 future 对象。而这个 future 对象必定是放在一个队列外面的。

下一大节，带大家去证实一下。

CompletionService 原理

首先 CompletionService 是一个接口：

ExecutorCompletionService 是这个接口的实现类：

看一下 ExecutorCompletionService 的构造方法：

能够看到是须要传入一个线程池对象的。队列默认应用的是 LinkedBlockingQueue。

当然，咱们也能够指定应用什么队列：

而后再看一下它的工作提交形式：

因为用 ExecutorCompletionService 次要是为了优雅的解决返回值。所以它反对两种 submit 类型的提交，都是有返回值的。

下面工夫治理巨匠版本海王应用的就是 Callable 类型的办法。

咱们先比照一下 Executor 间接提交和 ExecutorCompletionService 提交的差别：

差别就在 execute 办法外面。

ExecutorCompletionService 提交工作的时候是这样的：

`executor.execute(new QueueingFuture(f));
`

差别就在 execute 办法外面的 Runable：

看一下这个 QueueingFuture 是个什么货色：

机密基本上就在这个外面了。

QueueingFuture 继承自 FutureTask。重写了 done 办法，而后把 task 放到 queue 外面。

这个办法的含意就是当工作执行实现后，就会被放到队列外面去了。也就是说队列外面的 task 都是曾经 done 了的 task，而这个 task 就是一个个 future。

如果调用 queue 的 task 办法，就是阻塞期待。等到的肯定是就绪了的 future，调用 get 就能立马取得后果。

你说这一套操作是在干啥？

这不就是在做解耦吗？

之前你提交工作后还须要间接关怀每个工作返回的 future。当初 CompletionService 帮你对这些 future 进行了跟踪。

实现了调用者和 future 之间的解耦。

原理剖析完了，说一个须要留神的中央。

当你的应用场景是不关怀返回值的时候千万不要闲的蛋疼的用 CompletionService 去提交工作。

为什么？

因为后面说了，外面有个队列。而当你不关怀返回值的时候也就是不会去解决这个队列，导致这个队列外面的对象沉积的越来越多。

最初，炸了，OOM 了。

在开源框架中的利用

后面说了 CompletionService 是一个接口。除了 JDK 的 ExecutorCompletionService 实现了这个接口。

在开源框架外面也有相应的实现。比方 Redisson：

你去看这个实现，和 ExecutorCompletionService 思维是截然不同的，然而有些许的不一样。

它把 future 放到队列面的时候，没有重写 done 办法，而是应用了响应式编程的 onComplete：

而 CompletionService 的思维外围是：Executor 加 Queue。

这个思维，让我想起了在 Dubbo 中看到过的一段代码：

我已经在《Dubbo Cluster 集群那点你不晓得的事》这篇文章中提到过这个类。

这个类的 doInvoker 办法中的外围逻辑如下：

首先标号为 ① 的中央定义了一个队列。

标号为 ② 的中央在循环体中提交异步工作。有几个服务提供者就有几次循环。

子线程在标号为 ③ 的中央把返回后果放到队列外面。

只有一放进去，就能被标号为 ④ 的中央获取到（指定工夫内），而后程序立刻返回。

这样就能实现并行调用多个服务器，只有有一个服务器返回就立刻返回的性能。

我感觉这个思维和 CompletionService 的思维有一点点的相通之处的。

咱们要学 CompletionService，也要学它的思维。

最初说一句 (求关注)

嗯，写完了。这周是疯狂连轴旋转的一周，我在公司外面走动，都不是走路，都是在一路小跑的，停更一周的想法从周一就呈现了无数次。最初昨天加班加点的把公司安顿的“政治编程工作”实现之后，开始快马加鞭的筹备这篇文章，我还是怼进去了，毕竟我也是一个工夫治理巨匠。

下周就是间断周更文章一年整啦。过来的 365 天，过来的 52 周，我放弃了一周至多输入一篇优质原创文章的节奏。尽管没有多少粉丝（一年了都没过一万关注，真的是没脸见人），然而立下的 flag 算是超额完成了。

保持不住的时候再保持一下，的确是一种难能可贵的品质。

好啦，感谢您的浏览，我保持原创，非常欢送并感谢您的关注。

我是 why，一个被代码耽搁的文学创作者，不是大佬，然而喜爱分享，是一个又暖又有料的四川好男人。

退出移动版