关于java:送你一个并发编程的奇淫技巧要你舒服的起飞

工夫飞逝啊，我打算把本人过往的所有文章都梳理一遍。

不得不说，这真的是一个十分巨大的工程。

在整顿的过程中，我发现还是很多晚期写的作品，浏览量十分的拉胯。

在整顿文章的这个过程中，我发现其实有好多我之前写过的知识点，我本人可能都记不清楚了。

羞愧羞愧。

然而这的确是一个再次学习坚固的过程，挺好。

本文就来聊聊 CompletionService 这个货色。

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

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

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

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

有一篇文章外面写了啊：

因为是一组计算工作，你想拿到返回值去搞事件。这个返回值就被封装在 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 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();
    }
}

输入后果如下：

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

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

什么？你个中央空调还问我“什么是海王”？

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

先间接看用法：

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

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

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

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

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

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

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

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

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

先比照一下输入后果，整体起立，一起鼓掌：

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

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

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

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

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

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

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

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

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

首先 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 进行了跟踪。

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

为什么？

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

最初，炸了，OOM 了。

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

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

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

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

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

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

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

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

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

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

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

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

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

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

关于java:送你一个并发编程的奇淫技巧要你舒服的起飞

写在后面

再谈 Future

CompletionService 援救海王

CompletionService 原理

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

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

在开源框架中的利用

Just My Socks（注册教程内含优惠码）

关于java:送你一个并发编程的奇淫技巧要你舒服的起飞

写在后面

再谈 Future

CompletionService 援救海王

CompletionService 原理

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

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

在开源框架中的利用

Just My Socks（注册教程 内含优惠码）

Just My Socks（注册教程内含优惠码）