多线程使用不当导致的 OOM
Java之间
共 10325字，需浏览 21分钟
·
2022-07-10 16:37
往期热门文章： 1、老板要我开发一个简单的工作流引擎 2、Spring Boot 启动时自动执行代码的几种方式，还有谁不会？？ 3、Lombok原理和同时使⽤@Data和@Builder 的坑 4、如何用 Java 几分钟处理完 30 亿个数据? 5、面试官 | Spring Boot 项目如何统一结果，统一异常，统一日志？
事故描述
整体经过
事故根本原因
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();
}
探寻问题根源
ExecutorCompletionService 的 “套路”，我们用 ExecutorService 来作为对比，可以让我们更好地清楚什么场景下用 ExecutorCompletionService。ExecutorService 代码（建议下载后自己跑一跑）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();
}
}
第一步：主线程把三个任务提交到线程池里面去，把对应返回的 Future 放到 List 里面存起来，然后执行“都通知完了,等着接吧。”这行输出语句；
第二步：在循环里面执行
future.get()操作，阻塞等待。
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();
}
ExecutorCompletionService 使用了：completionService.take().get();
take() 然后再 get() 呢？CompletionService 接口以及接口的实现类1、ExecutorCompletionService 是 CompletionService 接口的实现类
2、接着跟一下 ExecutorCompletionService 的构造方法。
LinkedBlockingQueue，不过还有另外一个构造方法可以指定队列类型，如下两张图，有两个构造方法。默认 LinkedBlockingQueue 的构造方法。
3、Submit 任务提交的两种方式，都是有返回值的，我们例子中用到的就是第一种 Callable 类型的方法。
4、对比ExecutorService 和 ExecutorCompletionService 的 submit 方法可以看出区别。
5、差异就在 QueueingFuture。
QueueingFuture继承自FutureTask，而且红线部分标注的位置，重写了 done() 方法；把 task 放到
completionQueue队列里面。当任务执行完成后，task 就会被放到队列里面去了；此时此刻，
completionQueue队列里面的 task 都是已经done()完成了的 task。而这个 task 就是我们拿到的一个个的 future 结果；如果调用
completionQueue的 task 方法，会阻塞等待任务。等到的一定是完成了的 future，我们调用.get()方法 就能立马获得结果。
我们在使用
ExecutorService submit提交任务后需要关注每个任务返回的 future。然而CompletionService对这些 future 进行了追踪，并且重写了 done 方法，让你等的 completionQueue 队列里面一定是完成了的 task；作为网关 RPC 层，我们不用因为某一个接口的响应慢拖累所有的请求，可以在处理最快响应的业务场景里使用
CompletionService。
但是请注意！也是本次事故的核心问题。
ExecutorCompletionService 下面的 3 个方法的任意一个时，阻塞队列中的 task 执行结果才会从队列中移除掉，释放堆内存。
CompletionService。假如使用了，记得一定要从阻塞队列中移除掉 task 执行结果，避免 OOM！总结
上线前
严格的代码 review 习惯，一定要交给 back 人去看，毕竟自己写的代码自己是看不出问题的，相信每个程序猿都有这个自信；
上线记录：备注好上一个可回滚的包版本（给自己留一个后路）；
上线前确认回滚后，业务是否可降级。如果不可降级，一定要严格拉长这次上线的监控周期。
上线后
持续关注内存增长情况（这部分极容易被忽略，大家对内存的重视度不如 CPU 使用率）；
持续关注 CPU 使用率增长情况
GC 情况、线程数是否增长、是否有频繁的 Full GC 等；
关注服务性能报警，TP99、999 、MAX 是否出现明显的增高。
来源：juejin.cn/post/7064376361334358046
最近热文阅读：
1、面试官 | Spring Boot 项目如何统一结果，统一异常，统一日志？ 2、为什么不建议使用ON DUPLICATE KEY UPDATE？ 3、Java8 Stream，过分丝滑！ 4、8 种最坑SQL语法，工作中踩过吗？ 5、Java 语言“坑爹” TOP 10 6、你还不明白如何解决分布式Session？看这篇就够了！ 7、能解决 80% 故障的排查思路 8、程序员坐牢了，会被安排写代码吗？ 9、面试被问Nginx，怎么破? 10、为什么很多 SpringBoot 开发者放弃了 Tomcat，选择了 Undertow? 关注公众号，你想要的Java都在这里
评论