点击上方“芋道源码”,选择“设为星标”
管她前浪,还是后浪?
能浪的浪,才是好浪!
每天 10:33 更新文章,每天掉亿点点头发...
源码精品专栏
老规矩,我们先看下事故过程:某日,从 6 点 32 分开始少量用户访问 app 时会出现首页访问异常,到 7 点 20 分首页服务大规模不可用,7 点 36 分问题解决。
基于 Spring Boot + MyBatis Plus + Vue & Element 实现的后台管理系统 + 用户小程序,支持 RBAC 动态权限、多租户、数据权限、工作流、三方登录、支付、短信、商城等功能。 项目地址:https://github.com/YunaiV/ruoyi-vue-pro
事故的整个经过如下:
基于微服务的思想,构建在 B2C 电商场景下的项目实战。核心技术栈,是 Spring Boot + Dubbo 。未来,会重构成 Spring Cloud Alibaba 。 项目地址:https://github.com/YunaiV/onemall
事故代码模拟如下:
public static void test() throws InterruptedException, ExecutionException {
Executor executor = Executors.newFixedThreadPool(3);
CompletionService<String> service = new ExecutorCompletionService<>(executor);
service.submit(new Callable<String>() {
@Override
public String call() throws Exception {
return "HelloWorld--" + Thread.currentThread().getName();
}
});
}
先抛出问题,我们后面会详细阐述。问题的根源就在于 ExecutorCompletionService 结果没调用 take,poll 方法。
正确的写法如下所示:
public static void test() throws InterruptedException, ExecutionException {
Executor executor = Executors.newFixedThreadPool(3);
CompletionService<String> service = new ExecutorCompletionService<>(executor);
service.submit(new Callable<String>() {
@Override
public String call() throws Exception {
return "HelloWorld--" + Thread.currentThread().getName();
}
});
service.take().get();
}
一行代码引发的血案,而且不容易被发现,因为 OOM 是一个内存缓慢增长的过程,稍微粗心大意就会忽略,如果是这个代码块的调用量少的话,很可能几天甚至几个月后暴雷。
操作人回滚 or 重启服务器确实是最快的方式,但是如果不是事后快速分析出 OOM 的代码,而且不巧回滚的版本也是带 OOM 代码的,就比较悲催了。
如刚才所说,流量小了,回滚或者重启都可以释放内存;但是流量大的情况下,除非回滚到正常的版本,否则 GG。
接下来我们来探讨问题的根源,为了更好的理解 ExecutorCompletionService 的 “套路”,我们用 ExecutorService 来作为对比,可以让我们更好的清楚,什么场景下用 ExecutorCompletionService。
先看 ExecutorService 代码: (建议 down 下来跑一跑,以下代码建议吃饭的时候不要去看,味道略重!不过便于理解 orz)
public static void test1() throws Exception{
ExecutorService executorService = Executors.newCachedThreadPool();
ArrayList<Future<String>> futureArrayList = new ArrayList<>();
System.out.println("公司让你通知大家聚餐 你开车去接人");
Future<String> future10 = executorService.submit(() -> {
System.out.println("总裁:我在家上大号 我最近拉肚子比较慢 要蹲1个小时才能出来 你等会来接我吧");
TimeUnit.SECONDS.sleep(10);
System.out.println("总裁:1小时了 我上完大号了。你来接吧");
return "总裁上完大号了";
});
futureArrayList.add(future10);
Future<String> future3 = executorService.submit(() -> {
System.out.println("研发:我在家上大号 我比较快 要蹲3分钟就可以出来 你等会来接我吧");
TimeUnit.SECONDS.sleep(3);
System.out.println("研发:3分钟 我上完大号了。你来接吧");
return "研发上完大号了";
});
futureArrayList.add(future3);
Future<String> future6 = executorService.submit(() -> {
System.out.println("中层管理:我在家上大号 要蹲10分钟就可以出来 你等会来接我吧");
TimeUnit.SECONDS.sleep(6);
System.out.println("中层管理:10分钟 我上完大号了。你来接吧");
return "中层管理上完大号了";
});
futureArrayList.add(future6);
TimeUnit.SECONDS.sleep(1);
System.out.println("都通知完了,等着接吧。");
try {
for (Future<String> future : futureArrayList) {
String returnStr = future.get();
System.out.println(returnStr + ",你去接他");
}
Thread.currentThread().join();
} catch (Exception e) {
e.printStackTrace();
}
}
三个任务,每个任务执行时间分别是 10s、3s、6s。
通过 JDK 线程池的 submit 提交这三个 Callable 类型的任务:
最后结果如下:
先通知到总裁,也是先接总裁,足足等了 1 个小时,接到总裁后再去接研发和中层管理,尽管他们早就完事儿了,也得等总裁拉完~~
耗时最久的-10s 异步任务最先进入 list 执行,所以在循环过程中获取这个 10s 的任务结果的时候,get 操作会一直阻塞,直到 10s 异步任务执行完毕。即使 3s、5s 的任务早就执行完了,也得阻塞等待 10s 任务执行完。
看到这里,尤其是做网关业务的同学可能会产生共鸣,一般来说网关 RPC 会调用下游 N 多个接口,如下图:
如果都按照 ExecutorService 这种方式,并且恰巧前几个任务调用的接口耗时比较久,同时阻塞等待,那就比较悲催了。
所以 ExecutorCompletionService 应景而出。它作为任务线程的合理管控者,“任务规划师”的称号名副其实。
相同场景 ExecutorCompletionService 代码:
public static void test2() throws Exception {
ExecutorService executorService = Executors.newCachedThreadPool();
ExecutorCompletionService<String> completionService = new ExecutorCompletionService<>(executorService);
System.out.println("公司让你通知大家聚餐 你开车去接人");
completionService.submit(() -> {
System.out.println("总裁:我在家上大号 我最近拉肚子比较慢 要蹲1个小时才能出来 你等会来接我吧");
TimeUnit.SECONDS.sleep(10);
System.out.println("总裁:1小时了 我上完大号了。你来接吧");
return "总裁上完大号了";
});
completionService.submit(() -> {
System.out.println("研发:我在家上大号 我比较快 要蹲3分钟就可以出来 你等会来接我吧");
TimeUnit.SECONDS.sleep(3);
System.out.println("研发:3分钟 我上完大号了。你来接吧");
return "研发上完大号了";
});
completionService.submit(() -> {
System.out.println("中层管理:我在家上大号 要蹲10分钟就可以出来 你等会来接我吧");
TimeUnit.SECONDS.sleep(6);
System.out.println("中层管理:10分钟 我上完大号了。你来接吧");
return "中层管理上完大号了";
});
TimeUnit.SECONDS.sleep(1);
System.out.println("都通知完了,等着接吧。");
//提交了3个异步任务)
for (int i = 0; i < 3; i++) {
String returnStr = completionService.take().get();
System.out.println(returnStr + ",你去接他");
}
Thread.currentThread().join();
}
跑完结果如下:
这次就相对高效了一些,虽然先通知的总裁,但是根据大家上大号的速度,谁先拉完先去接谁,不用等待上大号最久的总裁了(现实生活里,建议采用第一种,不等总裁的后果 emmm 哈哈哈)。
放在一起对比下输出结果:
两段代码的差异非常小,获取结果的时候 ExecutorCompletionService 使用了:
completionService.take().get();
为毛要用 take() 然后再 get() 呢????我们看看源码:
ExecutorCompletionService 是 CompletionService 接口的实现类:
接着跟一下 ExecutorCompletionService 的构造方法,可以看到入参需要传一个线程池对象,默认使用的队列是 LinkedBlockingQueue,不过还有另外一个构造方法可以指定队列类型,如下两张图,两个构造方法。
默认 LinkedBlockingQueue 的构造方法:
可选队列类型的构造方法:
submit 任务提交的两种方式,都是有返回值的,我们例子中用到的就是第一种 Callable 类型的方法。
对比 ExecutorService 和 ExecutorCompletionService submit 方法,可以看出区别。
ExecutorService:
ExecutorCompletionService:
差异就在 QueueingFuture,这个到底作用是啥?
我们继续跟进去看:
看到这里,相信大家伙都应该多少明白点了:
当只有调用了 ExecutorCompletionService 下面的 3 个方法的任意一个时,阻塞队列中的 task 执行结果才会从队列中移除掉,释放堆内存。
由于该业务不需要使用任务的返回值,则没进行调用 take,poll 方法。从而导致没有释放堆内存,堆内存会随着调用量的增加一直增长。
所以,业务场景中不需要使用任务返回值的 别没事儿使用 CompletionService,假如使用了,记得一定要从阻塞队列中移除掉 task 执行结果,避免 OOM!
知道事故的原因,我们来总结下方法论,毕竟孔子他老人家说过:自省吾身,常思己过,善修其身!
上线前:
上线后:
欢迎加入我的知识星球,一起探讨架构,交流源码。加入方式,长按下方二维码噢:
已在知识星球更新源码解析如下:
最近更新《芋道 SpringBoot 2.X 入门》系列,已经 101 余篇,覆盖了 MyBatis、Redis、MongoDB、ES、分库分表、读写分离、SpringMVC、Webflux、权限、WebSocket、Dubbo、RabbitMQ、RocketMQ、Kafka、性能测试等等内容。
提供近 3W 行代码的 SpringBoot 示例,以及超 4W 行代码的电商微服务项目。
获取方式:点“在看”,关注公众号并回复 666 领取,更多内容陆续奉上。
文章有帮助的话,在看,转发吧。谢谢支持哟 (*^__^*)