外贸开发模板网站模板,葫芦岛市网站建设,瀚钦科技网站建设,中石化十建公司官网1、背景
在我们的项目中有这么一个场景#xff0c;需要消费kafka中的消息#xff0c;并生成对应的工单数据。早些时候程序运行的好好的#xff0c;但是有一天#xff0c;我们升级了容器的配置#xff0c;结果导致部分消息无法消费。而消费者的代码是使用CompletableFutur…1、背景
在我们的项目中有这么一个场景需要消费kafka中的消息并生成对应的工单数据。早些时候程序运行的好好的但是有一天我们升级了容器的配置结果导致部分消息无法消费。而消费者的代码是使用CompletableFuture.runAsync(() - {while (true){ ..... }}) 来实现的。 即
需要消费Kafka topic的个数 7个每个线程消费一个topic消费方式使用线程池异步消费消费池默认的 ForkJoin 线程池???并且没有做任何配置是否会释放线程池中的核心线程 不会释放没出问题时容器配置 2核4G出问题时容器配置4核8G影响的结果只有3个topic的数据可以消费。
2、容器2核4G可以正常消费 即此时程序会启动7个线程来进行消费。
3、容器4核8G只有部分可以消费 即此时程序会启动3个线程来进行消费。
4、问题原因分析
1、通过上面的背景我们可以知道是因为升级了容器的配置才导致我们消费kafka中的消息失败了。 2、针对kafka中的每个topic我们都会使用一个单独的线程来消费并且不会释放这个线程。 3、而线程的启动方式是通过CompletableFuture.runAsync()方法来启动的那么通过这种方式启动的线程是每个任务一个启动一个线程还是只启动固定的线程呢.
通过以上分析那么问题肯定是出现在线程池身上那么我们默认使用的是什么线程池呢查看CompletableFuture.runAsync()的源码可知有一定的几率是ForkJoinPool。那么我们一起看下源码。
5、源码分析 1、确认使用什么线程池
public static CompletableFutureVoid runAsync(Runnable runnable) {return asyncRunStage(asyncPool, runnable);
}
private static final Executor asyncPool useCommonPool ?ForkJoinPool.commonPool() : new ThreadPerTaskExecutor();通过上述源码可知我们可能使用的ForkJoin线程池也可能使用的是ThreadPerTaskExecutor线程池。
ThreadPerTaskExecutor 这个是每个任务一个线程。ForkJoinPool 那么就需要确定启动了多少个线程。
2、确认是否使用 ForkJoin 线程池
需要确定 useCommonPool 字段是如何赋值的。
private static final boolean useCommonPool (ForkJoinPool.getCommonPoolParallelism() 1);通过上面代码可知是否使用ForkJoin线程池是由 ForkJoinPool.getCommonPoolParallelism()的值确定的。即并行度是否大于1大于则使用ForkJoin线程池
public static int getCommonPoolParallelism() {return commonParallelism;
}3、commonParallelism 的赋值 1、从上图中可知parallelism的设置有2种方式
通过Jvm的启动参数java.util.concurrent.ForkJoinPool.common.parallelism进行设置且这个值最大为 MAX_CAP即32727。若没有通过Jvm的参数配置则有2种情况若cpu的核数1则返回1否则返回cpu的核数-1
2、commonParallelism的取值
common java.security.AccessController.doPrivileged(new java.security.PrivilegedActionForkJoinPool() {public ForkJoinPool run() { return makeCommonPool(); }});
int par common.config SMASK; // report 1 even if threads disabled
commonParallelism par 0 ? par : 1;SMASK 的值是 65535。 common.config 的值就是 (parallelism SMASK) | 0的值即最大为65535若parallelism的值为0则返回0。 int par common.config SMASK 即最大为 65535 commonParallelism par 0 ? par : 1 的值就为 parallelism的值或1
6、结论 结论 由上面的知识点我们可以得出当我们的容器是2核4G时程序选择的线程池是ThreadPerTaskExecutor当我们的容器是4核8G时程序选择的线程池是ForkJoinPool。
