From feb03aede64f6ab1f33430ccf90039700d439a2e Mon Sep 17 00:00:00 2001 From: zengqiao Date: Mon, 28 Nov 2022 15:10:32 +0800 Subject: [PATCH] =?UTF-8?q?[Optimize]=E4=BC=98=E5=8C=96=E7=BA=BF=E7=A8=8B?= =?UTF-8?q?=E6=B1=A0=E7=9A=84=E5=90=8D=E7=A7=B0(#789)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../km/collector/service/CollectThreadPoolService.java | 2 +- .../know/streaming/km/common/utils/FutureNoWaitUtil.java | 6 +++--- .../know/streaming/km/common/utils/FutureUtil.java | 6 +++--- .../know/streaming/km/common/utils/FutureWaitUtil.java | 4 ++-- .../streaming/km/task/service/TaskThreadPoolService.java | 6 +++--- 5 files changed, 12 insertions(+), 12 deletions(-) diff --git a/km-collector/src/main/java/com/xiaojukeji/know/streaming/km/collector/service/CollectThreadPoolService.java b/km-collector/src/main/java/com/xiaojukeji/know/streaming/km/collector/service/CollectThreadPoolService.java index 22505b35..f5d3c496 100644 --- a/km-collector/src/main/java/com/xiaojukeji/know/streaming/km/collector/service/CollectThreadPoolService.java +++ b/km-collector/src/main/java/com/xiaojukeji/know/streaming/km/collector/service/CollectThreadPoolService.java @@ -237,7 +237,7 @@ public class CollectThreadPoolService { private synchronized FutureWaitUtil closeOldAndCreateNew(Long shardId) { // 新的 FutureWaitUtil newFutureUtil = FutureWaitUtil.init( - "CollectorMetricsFutureUtil-Shard-" + shardId, + "MetricCollect-Shard-" + shardId, this.futureUtilThreadNum, this.futureUtilThreadNum, this.futureUtilQueueSize diff --git a/km-common/src/main/java/com/xiaojukeji/know/streaming/km/common/utils/FutureNoWaitUtil.java b/km-common/src/main/java/com/xiaojukeji/know/streaming/km/common/utils/FutureNoWaitUtil.java index 6bd7ae19..378abaf6 100644 --- a/km-common/src/main/java/com/xiaojukeji/know/streaming/km/common/utils/FutureNoWaitUtil.java +++ b/km-common/src/main/java/com/xiaojukeji/know/streaming/km/common/utils/FutureNoWaitUtil.java @@ -23,7 +23,7 @@ public class FutureNoWaitUtil { private FutureNoWaitUtil() { } - public static FutureNoWaitUtil init(String name, int corePoolSize, int maxPoolSize, int queueSize) { + public static FutureNoWaitUtil init(String threadPoolName, int corePoolSize, int maxPoolSize, int queueSize) { FutureNoWaitUtil futureUtil = new FutureNoWaitUtil<>(); // 创建任务线程池 @@ -33,7 +33,7 @@ public class FutureNoWaitUtil { 300, TimeUnit.SECONDS, new LinkedBlockingDeque<>(queueSize), - new NamedThreadFactory("KS-KM-FutureNoWaitUtil-" + name), + new NamedThreadFactory(threadPoolName), new ThreadPoolExecutor.DiscardOldestPolicy() //对拒绝任务不抛弃，而是抛弃队列里面等待最久的一个线程，然后把拒绝任务加到队列。 ); futureUtil.executor.allowCoreThreadTimeOut(true); @@ -41,7 +41,7 @@ public class FutureNoWaitUtil { futureUtil.delayQueueData = new DelayQueue<>(); // 创建检查延迟队列的线程并启动 - futureUtil.checkDelayQueueThread = new Thread(() -> futureUtil.runCheck(), "KS-KM-FutureNoWaitUtil-CheckDelayQueueData-" + name); + futureUtil.checkDelayQueueThread = new Thread(() -> futureUtil.runCheck(), threadPoolName + "-CheckTaskTimeout"); futureUtil.checkDelayQueueThread.setDaemon(true); futureUtil.checkDelayQueueThread.start(); diff --git a/km-common/src/main/java/com/xiaojukeji/know/streaming/km/common/utils/FutureUtil.java b/km-common/src/main/java/com/xiaojukeji/know/streaming/km/common/utils/FutureUtil.java index 1e5cd875..ed68939f 100644 --- a/km-common/src/main/java/com/xiaojukeji/know/streaming/km/common/utils/FutureUtil.java +++ b/km-common/src/main/java/com/xiaojukeji/know/streaming/km/common/utils/FutureUtil.java @@ -9,12 +9,12 @@ import java.util.concurrent.*; public class FutureUtil { private ThreadPoolExecutor executor; - public static final FutureUtil quickStartupFutureUtil = FutureUtil.init("QuickStartupFutureUtil", 8, 8, 10240); + public static final FutureUtil quickStartupFutureUtil = FutureUtil.init("QuickStartupTP", 8, 8, 10240); private FutureUtil() { } - public static FutureUtil init(String name, int corePoolSize, int maxPoolSize, int queueSize) { + public static FutureUtil init(String threadPoolName, int corePoolSize, int maxPoolSize, int queueSize) { FutureUtil futureUtil = new FutureUtil<>(); futureUtil.executor = new ThreadPoolExecutor( @@ -23,7 +23,7 @@ public class FutureUtil { 300, TimeUnit.SECONDS, new LinkedBlockingDeque<>(queueSize), - new NamedThreadFactory("FutureUtil-" + name), + new NamedThreadFactory(threadPoolName), new ThreadPoolExecutor.DiscardOldestPolicy() //对拒绝任务不抛弃，而是抛弃队列里面等待最久的一个线程，然后把拒绝任务加到队列。 ); futureUtil.executor.allowCoreThreadTimeOut(true); diff --git a/km-common/src/main/java/com/xiaojukeji/know/streaming/km/common/utils/FutureWaitUtil.java b/km-common/src/main/java/com/xiaojukeji/know/streaming/km/common/utils/FutureWaitUtil.java index cf2af736..e62efb8b 100644 --- a/km-common/src/main/java/com/xiaojukeji/know/streaming/km/common/utils/FutureWaitUtil.java +++ b/km-common/src/main/java/com/xiaojukeji/know/streaming/km/common/utils/FutureWaitUtil.java @@ -23,7 +23,7 @@ public class FutureWaitUtil { private FutureWaitUtil() { } - public static FutureWaitUtil init(String name, int corePoolSize, int maxPoolSize, int queueSize) { + public static FutureWaitUtil init(String threadPoolName, int corePoolSize, int maxPoolSize, int queueSize) { FutureWaitUtil futureUtil = new FutureWaitUtil<>(); futureUtil.executor = new ThreadPoolExecutor( @@ -32,7 +32,7 @@ public class FutureWaitUtil { 300, TimeUnit.SECONDS, new LinkedBlockingDeque<>(queueSize), - new NamedThreadFactory("FutureWaitUtil-" + name), + new NamedThreadFactory(threadPoolName), new ThreadPoolExecutor.DiscardOldestPolicy() //对拒绝任务不抛弃，而是抛弃队列里面等待最久的一个线程，然后把拒绝任务加到队列。 ); futureUtil.executor.allowCoreThreadTimeOut(true); diff --git a/km-task/src/main/java/com/xiaojukeji/know/streaming/km/task/service/TaskThreadPoolService.java b/km-task/src/main/java/com/xiaojukeji/know/streaming/km/task/service/TaskThreadPoolService.java index f4bb6826..33545f04 100644 --- a/km-task/src/main/java/com/xiaojukeji/know/streaming/km/task/service/TaskThreadPoolService.java +++ b/km-task/src/main/java/com/xiaojukeji/know/streaming/km/task/service/TaskThreadPoolService.java @@ -52,21 +52,21 @@ public class TaskThreadPoolService { @PostConstruct private void init() { metricsTaskThreadPool = FutureNoWaitUtil.init( - "metricsTaskThreadPool", + "MetricsTaskTP", metricsTaskThreadNum, metricsTaskThreadNum, metricsTaskQueueSize ); metadataTaskThreadPool = FutureNoWaitUtil.init( - "metadataTaskThreadPool", + "MetadataTaskTP", metadataTaskThreadNum, metadataTaskThreadNum, metadataTaskQueueSize ); commonTaskThreadPool = FutureNoWaitUtil.init( - "commonTaskThreadPool", + "CommonTaskTP", commonTaskThreadNum, commonTaskThreadNum, commonTaskQueueSize