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

4次阅读

共计 6058 个字符,预计需要花费 16 分钟才能阅读完成。

写在后面

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

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

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

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

羞愧羞愧。

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

再谈 Future

本文就来聊聊 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 援救海王

还是下面的场景,当咱们引入了 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 原理

首先 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 中看到过的一个类:

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

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

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

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

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

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

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

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

正文完
 0