From 703d685d59c5f7961fedc943a9c5f319dc6c9753 Mon Sep 17 00:00:00 2001 From: zengqiao Date: Thu, 8 Sep 2022 14:17:15 +0800 Subject: [PATCH] =?UTF-8?q?Task=E4=BB=BB=E5=8A=A1=E5=88=86=E4=B8=BAmetrics?= =?UTF-8?q?,common,metaddata=E4=B8=89=E7=B1=BB=EF=BC=8C=E6=AF=8F=E4=B8=80?= =?UTF-8?q?=E7=B1=BB=E4=BB=BB=E5=8A=A1=E7=9A=84=E6=89=A7=E8=A1=8C=E5=AF=B9?= =?UTF-8?q?=E5=BA=94=E4=B8=80=E4=B8=AA=E7=BA=BF=E7=A8=8B=E6=B1=A0=EF=BC=8C?= =?UTF-8?q?=E5=87=8F=E5=B0=91=E5=AF=B9Job=E6=A8=A1=E5=9D=97=E7=BA=BF?= =?UTF-8?q?=E7=A8=8B=E6=B1=A0=E7=9A=84=E4=BE=9D=E8=B5=96?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- km-rest/src/main/resources/application.yml | 12 +++- .../task/AbstractAsyncCommonDispatchTask.java | 46 +++++++++++++ .../km/task/AbstractDispatchTask.java | 8 +++ .../km/task/client/CheckJmxClientTask.java | 2 +- .../km/task/health/HealthCheckTask.java | 22 ++---- .../km/task/job/CommunityReassignJobTask.java | 6 +- .../know/streaming/km/task/job/KMJobTask.java | 6 +- .../AbstractAsyncMetadataDispatchTask.java | 47 +++++++++++++ .../metadata/SyncBrokerConfigDiffTask.java | 7 +- .../km/task/metadata/SyncBrokerTask.java | 7 +- .../km/task/metadata/SyncControllerTask.java | 7 +- .../km/task/metadata/SyncKafkaAclTask.java | 7 +- .../km/task/metadata/SyncKafkaGroupTask.java | 7 +- .../km/task/metadata/SyncKafkaUserTask.java | 7 +- .../km/task/metadata/SyncPartitionTask.java | 7 +- .../km/task/metadata/SyncTopicConfigTask.java | 7 +- .../km/task/metadata/SyncTopicTask.java | 7 +- .../AbstractAsyncMetricsDispatchTask.java | 47 +++++++++++++ .../metrics/BrokerMetricCollectorTask.java | 17 ++--- .../metrics/ClusterMetricCollectorTask.java | 17 ++--- .../metrics/GroupMetricCollectorTask.java | 17 ++--- .../metrics/PartitionMetricCollectorTask.java | 17 ++--- .../metrics/ReplicaMetricCollectorTask.java | 17 ++--- .../metrics/TopicMetricCollectorTask.java | 15 +---- .../task/service/TaskThreadPoolService.java | 67 ++++++++++++++++--- 25 files changed, 279 insertions(+), 147 deletions(-) create mode 100644 km-task/src/main/java/com/xiaojukeji/know/streaming/km/task/AbstractAsyncCommonDispatchTask.java create mode 100644 km-task/src/main/java/com/xiaojukeji/know/streaming/km/task/metadata/AbstractAsyncMetadataDispatchTask.java create mode 100644 km-task/src/main/java/com/xiaojukeji/know/streaming/km/task/metrics/AbstractAsyncMetricsDispatchTask.java diff --git a/km-rest/src/main/resources/application.yml b/km-rest/src/main/resources/application.yml index 7af15333..08cac4af 100644 --- a/km-rest/src/main/resources/application.yml +++ b/km-rest/src/main/resources/application.yml @@ -60,9 +60,15 @@ thread-pool: suitable-queue-size: 1000 # 线程池理想的队列大小，非主要，可不修改 task: # 任务模块的配置 - heaven: # 采集任务配置 - thread-num: 20 # 采集任务线程池核心线程数 - queue-size: 1000 # 采集任务线程池队列大小 + metrics: # metrics采集任务配置 + thread-num: 18 # metrics采集任务线程池核心线程数 + queue-size: 180 # metrics采集任务线程池队列大小 + metadata: # metadata同步任务配置 + thread-num: 27 # metadata同步任务线程池核心线程数 + queue-size: 270 # metadata同步任务线程池队列大小 + common: # 剩余其他任务配置 + thread-num: 15 # 剩余其他任务线程池核心线程数 + queue-size: 150 # 剩余其他任务线程池队列大小 # 客户端池大小相关配置 diff --git a/km-task/src/main/java/com/xiaojukeji/know/streaming/km/task/AbstractAsyncCommonDispatchTask.java b/km-task/src/main/java/com/xiaojukeji/know/streaming/km/task/AbstractAsyncCommonDispatchTask.java new file mode 100644 index 00000000..0f8f5797 --- /dev/null +++ b/km-task/src/main/java/com/xiaojukeji/know/streaming/km/task/AbstractAsyncCommonDispatchTask.java @@ -0,0 +1,46 @@ +package com.xiaojukeji.know.streaming.km.task; + +import com.didiglobal.logi.job.common.TaskResult; +import com.didiglobal.logi.log.ILog; +import com.didiglobal.logi.log.LogFactory; +import com.xiaojukeji.know.streaming.km.common.bean.entity.cluster.ClusterPhy; +import com.xiaojukeji.know.streaming.km.task.service.TaskThreadPoolService; +import org.springframework.beans.factory.annotation.Autowired; + +/** + * other相关任务 + */ +public abstract class AbstractAsyncCommonDispatchTask extends AbstractClusterPhyDispatchTask { + private static final ILog log = LogFactory.getLog(AbstractAsyncCommonDispatchTask.class); + + public abstract TaskResult processClusterTask(ClusterPhy clusterPhy, long triggerTimeUnitMs) throws Exception; + + @Autowired + private TaskThreadPoolService taskThreadPoolService; + + @Override + protected TaskResult processSubTask(ClusterPhy clusterPhy, long triggerTimeUnitMs) throws Exception { + return this.asyncProcessSubTask(clusterPhy, triggerTimeUnitMs); + } + + public TaskResult asyncProcessSubTask(ClusterPhy clusterPhy, long triggerTimeUnitMs) { + taskThreadPoolService.submitCommonTask( + String.format("taskName=%s||clusterPhyId=%d", this.taskName, clusterPhy.getId()), + this.timeoutUnitSec.intValue() * 1000, + () -> { + try { + TaskResult tr = this.processClusterTask(clusterPhy, triggerTimeUnitMs); + if (TaskResult.SUCCESS_CODE != tr.getCode()) { + log.error("class=AbstractAsyncCommonDispatchTask||taskName={}||clusterPhyId={}||taskResult={}||msg=failed", this.taskName, clusterPhy.getId(), tr); + } else { + log.debug("class=AbstractAsyncCommonDispatchTask||taskName={}||clusterPhyId={}||msg=success", this.taskName, clusterPhy.getId()); + } + } catch (Exception e) { + log.error("class=AbstractAsyncCommonDispatchTask||taskName={}||clusterPhyId={}||errMsg=exception", this.taskName, clusterPhy.getId(), e); + } + } + ); + + return TaskResult.SUCCESS; + } +} diff --git a/km-task/src/main/java/com/xiaojukeji/know/streaming/km/task/AbstractDispatchTask.java b/km-task/src/main/java/com/xiaojukeji/know/streaming/km/task/AbstractDispatchTask.java index 0be382fe..6fce1bb3 100644 --- a/km-task/src/main/java/com/xiaojukeji/know/streaming/km/task/AbstractDispatchTask.java +++ b/km-task/src/main/java/com/xiaojukeji/know/streaming/km/task/AbstractDispatchTask.java @@ -130,4 +130,12 @@ public abstract class AbstractDispatchTask this.calAndUpdateHealthCheckResult(clusterPhy, triggerTimeUnitMs) - ); - - return TaskResult.SUCCESS; + public TaskResult processClusterTask(ClusterPhy clusterPhy, long triggerTimeUnitMs) { + return this.calAndUpdateHealthCheckResult(clusterPhy, triggerTimeUnitMs); } - private void calAndUpdateHealthCheckResult(ClusterPhy clusterPhy, long triggerTimeUnitMs) { + private TaskResult calAndUpdateHealthCheckResult(ClusterPhy clusterPhy, long triggerTimeUnitMs) { // 获取配置，<配置名，配置信息> Map healthConfigMap = healthCheckResultService.getClusterHealthConfig(clusterPhy.getId()); @@ -91,6 +81,8 @@ public class HealthCheckTask extends AbstractClusterPhyDispatchTask { } catch (Exception e) { log.error("method=processSubTask||clusterPhyId={}||errMsg=exception!", clusterPhy.getId(), e); } + + return TaskResult.SUCCESS; } private List getNoResResult(Long clusterPhyId, AbstractHealthCheckService healthCheckService, Map healthConfigMap) { diff --git a/km-task/src/main/java/com/xiaojukeji/know/streaming/km/task/job/CommunityReassignJobTask.java b/km-task/src/main/java/com/xiaojukeji/know/streaming/km/task/job/CommunityReassignJobTask.java index 179aa30a..3a01e80c 100644 --- a/km-task/src/main/java/com/xiaojukeji/know/streaming/km/task/job/CommunityReassignJobTask.java +++ b/km-task/src/main/java/com/xiaojukeji/know/streaming/km/task/job/CommunityReassignJobTask.java @@ -8,7 +8,7 @@ import com.didiglobal.logi.log.LogFactory; import com.xiaojukeji.know.streaming.km.common.bean.entity.cluster.ClusterPhy; import com.xiaojukeji.know.streaming.km.common.bean.entity.result.Result; import com.xiaojukeji.know.streaming.km.core.service.reassign.ReassignJobService; -import com.xiaojukeji.know.streaming.km.task.AbstractClusterPhyDispatchTask; +import com.xiaojukeji.know.streaming.km.task.AbstractAsyncCommonDispatchTask; import org.springframework.beans.factory.annotation.Autowired; @Task(name = "CommunityReassignJobTask", @@ -17,14 +17,14 @@ import org.springframework.beans.factory.annotation.Autowired; autoRegister = true, consensual = ConsensualEnum.BROADCAST, timeout = 6 * 60) -public class CommunityReassignJobTask extends AbstractClusterPhyDispatchTask { +public class CommunityReassignJobTask extends AbstractAsyncCommonDispatchTask { private static final ILog log = LogFactory.getLog(CommunityReassignJobTask.class); @Autowired private ReassignJobService reassignJobService; @Override - protected TaskResult processSubTask(ClusterPhy clusterPhy, long triggerTimeUnitMs) throws Exception { + public TaskResult processClusterTask(ClusterPhy clusterPhy, long triggerTimeUnitMs) throws Exception { // 获取迁移中的任务 Long jobId = reassignJobService.getOneRunningJobId(clusterPhy.getId()); if (jobId == null) { diff --git a/km-task/src/main/java/com/xiaojukeji/know/streaming/km/task/job/KMJobTask.java b/km-task/src/main/java/com/xiaojukeji/know/streaming/km/task/job/KMJobTask.java index 3d3039b8..c0b64e63 100644 --- a/km-task/src/main/java/com/xiaojukeji/know/streaming/km/task/job/KMJobTask.java +++ b/km-task/src/main/java/com/xiaojukeji/know/streaming/km/task/job/KMJobTask.java @@ -5,7 +5,7 @@ import com.didiglobal.logi.job.common.TaskResult; import com.didiglobal.logi.job.core.consensual.ConsensualEnum; import com.xiaojukeji.know.streaming.km.common.bean.entity.cluster.ClusterPhy; import com.xiaojukeji.know.streaming.km.core.service.job.JobService; -import com.xiaojukeji.know.streaming.km.task.AbstractClusterPhyDispatchTask; +import com.xiaojukeji.know.streaming.km.task.AbstractAsyncCommonDispatchTask; import org.springframework.beans.factory.annotation.Autowired; @Task(name = "kmJobTask", @@ -14,13 +14,13 @@ import org.springframework.beans.factory.annotation.Autowired; autoRegister = true, consensual = ConsensualEnum.BROADCAST, timeout = 6 * 60) -public class KMJobTask extends AbstractClusterPhyDispatchTask { +public class KMJobTask extends AbstractAsyncCommonDispatchTask { @Autowired private JobService jobService; @Override - protected TaskResult processSubTask(ClusterPhy clusterPhy, long triggerTimeUnitMs) throws Exception { + public TaskResult processClusterTask(ClusterPhy clusterPhy, long triggerTimeUnitMs) throws Exception { jobService.scheduleJobByClusterId(clusterPhy.getId()); return TaskResult.SUCCESS; } diff --git a/km-task/src/main/java/com/xiaojukeji/know/streaming/km/task/metadata/AbstractAsyncMetadataDispatchTask.java b/km-task/src/main/java/com/xiaojukeji/know/streaming/km/task/metadata/AbstractAsyncMetadataDispatchTask.java new file mode 100644 index 00000000..668208c7 --- /dev/null +++ b/km-task/src/main/java/com/xiaojukeji/know/streaming/km/task/metadata/AbstractAsyncMetadataDispatchTask.java @@ -0,0 +1,47 @@ +package com.xiaojukeji.know.streaming.km.task.metadata; + +import com.didiglobal.logi.job.common.TaskResult; +import com.didiglobal.logi.log.ILog; +import com.didiglobal.logi.log.LogFactory; +import com.xiaojukeji.know.streaming.km.common.bean.entity.cluster.ClusterPhy; +import com.xiaojukeji.know.streaming.km.task.AbstractClusterPhyDispatchTask; +import com.xiaojukeji.know.streaming.km.task.service.TaskThreadPoolService; +import org.springframework.beans.factory.annotation.Autowired; + +/** + * 元数据同步相关任务 + */ +public abstract class AbstractAsyncMetadataDispatchTask extends AbstractClusterPhyDispatchTask { + private static final ILog log = LogFactory.getLog(AbstractAsyncMetadataDispatchTask.class); + + public abstract TaskResult processClusterTask(ClusterPhy clusterPhy, long triggerTimeUnitMs) throws Exception; + + @Autowired + private TaskThreadPoolService taskThreadPoolService; + + @Override + protected TaskResult processSubTask(ClusterPhy clusterPhy, long triggerTimeUnitMs) throws Exception { + return this.asyncProcessSubTask(clusterPhy, triggerTimeUnitMs); + } + + public TaskResult asyncProcessSubTask(ClusterPhy clusterPhy, long triggerTimeUnitMs) { + taskThreadPoolService.submitMetadataTask( + String.format("taskName=%s||clusterPhyId=%d", this.taskName, clusterPhy.getId()), + this.timeoutUnitSec.intValue() * 1000, + () -> { + try { + TaskResult tr = this.processClusterTask(clusterPhy, triggerTimeUnitMs); + if (TaskResult.SUCCESS_CODE != tr.getCode()) { + log.error("class=AbstractAsyncMetadataDispatchTask||taskName={}||clusterPhyId={}||taskResult={}||msg=failed", this.taskName, clusterPhy.getId(), tr); + } else { + log.debug("class=AbstractAsyncMetadataDispatchTask||taskName={}||clusterPhyId={}||msg=success", this.taskName, clusterPhy.getId()); + } + } catch (Exception e) { + log.error("class=AbstractAsyncMetadataDispatchTask||taskName={}||clusterPhyId={}||errMsg=exception", this.taskName, clusterPhy.getId(), e); + } + } + ); + + return TaskResult.SUCCESS; + } +} diff --git a/km-task/src/main/java/com/xiaojukeji/know/streaming/km/task/metadata/SyncBrokerConfigDiffTask.java b/km-task/src/main/java/com/xiaojukeji/know/streaming/km/task/metadata/SyncBrokerConfigDiffTask.java index ea286d3e..60ffbd84 100644 --- a/km-task/src/main/java/com/xiaojukeji/know/streaming/km/task/metadata/SyncBrokerConfigDiffTask.java +++ b/km-task/src/main/java/com/xiaojukeji/know/streaming/km/task/metadata/SyncBrokerConfigDiffTask.java @@ -13,7 +13,6 @@ import com.xiaojukeji.know.streaming.km.common.bean.po.broker.BrokerConfigPO; import com.xiaojukeji.know.streaming.km.common.enums.config.ConfigDiffTypeEnum; import com.xiaojukeji.know.streaming.km.core.service.broker.BrokerConfigService; import com.xiaojukeji.know.streaming.km.core.service.broker.BrokerService; -import com.xiaojukeji.know.streaming.km.task.AbstractClusterPhyDispatchTask; import org.springframework.beans.factory.annotation.Autowired; import java.util.*; @@ -25,12 +24,12 @@ import java.util.stream.Collectors; * @date 22/02/25 */ @Task(name = "SyncBrokerConfigDiffTask", - description = "Broker配置的Diff信息同步到DB,", + description = "Broker配置的Diff信息同步到DB", cron = "0 0/1 * * * ? *", autoRegister = true, consensual = ConsensualEnum.BROADCAST, timeout = 6 * 60) -public class SyncBrokerConfigDiffTask extends AbstractClusterPhyDispatchTask { +public class SyncBrokerConfigDiffTask extends AbstractAsyncMetadataDispatchTask { protected static final ILog log = LogFactory.getLog(SyncBrokerConfigDiffTask.class); @Autowired @@ -40,7 +39,7 @@ public class SyncBrokerConfigDiffTask extends AbstractClusterPhyDispatchTask { private BrokerConfigService brokerConfigService; @Override - public TaskResult processSubTask(ClusterPhy clusterPhy, long triggerTimeUnitMs) { + public TaskResult processClusterTask(ClusterPhy clusterPhy, long triggerTimeUnitMs) { // > Map> allConfigMap = new HashMap<>(); diff --git a/km-task/src/main/java/com/xiaojukeji/know/streaming/km/task/metadata/SyncBrokerTask.java b/km-task/src/main/java/com/xiaojukeji/know/streaming/km/task/metadata/SyncBrokerTask.java index 78ff9560..a4d1dc99 100644 --- a/km-task/src/main/java/com/xiaojukeji/know/streaming/km/task/metadata/SyncBrokerTask.java +++ b/km-task/src/main/java/com/xiaojukeji/know/streaming/km/task/metadata/SyncBrokerTask.java @@ -9,25 +9,24 @@ import com.xiaojukeji.know.streaming.km.common.bean.entity.broker.Broker; import com.xiaojukeji.know.streaming.km.common.bean.entity.cluster.ClusterPhy; import com.xiaojukeji.know.streaming.km.common.bean.entity.result.Result; import com.xiaojukeji.know.streaming.km.core.service.broker.BrokerService; -import com.xiaojukeji.know.streaming.km.task.AbstractClusterPhyDispatchTask; import org.springframework.beans.factory.annotation.Autowired; import java.util.List; @Task(name = "SyncBrokerTask", - description = "Broker信息同步到DB,", + description = "Broker信息同步到DB", cron = "0 0/1 * * * ? *", autoRegister = true, consensual = ConsensualEnum.BROADCAST, timeout = 2 * 60) -public class SyncBrokerTask extends AbstractClusterPhyDispatchTask { +public class SyncBrokerTask extends AbstractAsyncMetadataDispatchTask { private static final ILog log = LogFactory.getLog(SyncBrokerTask.class); @Autowired private BrokerService brokerService; @Override - public TaskResult processSubTask(ClusterPhy clusterPhy, long triggerTimeUnitMs) { + public TaskResult processClusterTask(ClusterPhy clusterPhy, long triggerTimeUnitMs) { Result> brokersResult = brokerService.listBrokersFromKafka(clusterPhy); if (brokersResult.failed()) { return new TaskResult(TaskResult.FAIL_CODE, brokersResult.getMessage()); diff --git a/km-task/src/main/java/com/xiaojukeji/know/streaming/km/task/metadata/SyncControllerTask.java b/km-task/src/main/java/com/xiaojukeji/know/streaming/km/task/metadata/SyncControllerTask.java index e87122a6..971ef852 100644 --- a/km-task/src/main/java/com/xiaojukeji/know/streaming/km/task/metadata/SyncControllerTask.java +++ b/km-task/src/main/java/com/xiaojukeji/know/streaming/km/task/metadata/SyncControllerTask.java @@ -9,24 +9,23 @@ import com.xiaojukeji.know.streaming.km.common.bean.entity.cluster.ClusterPhy; import com.xiaojukeji.know.streaming.km.common.bean.entity.kafkacontroller.KafkaController; import com.xiaojukeji.know.streaming.km.common.bean.entity.result.Result; import com.xiaojukeji.know.streaming.km.core.service.kafkacontroller.KafkaControllerService; -import com.xiaojukeji.know.streaming.km.task.AbstractClusterPhyDispatchTask; import org.springframework.beans.factory.annotation.Autowired; @Task(name = "SyncControllerTask", - description = "Controller信息同步到DB,", + description = "Controller信息同步到DB", cron = "0 0/1 * * * ? *", autoRegister = true, consensual = ConsensualEnum.BROADCAST, timeout = 2 * 60) -public class SyncControllerTask extends AbstractClusterPhyDispatchTask { +public class SyncControllerTask extends AbstractAsyncMetadataDispatchTask { private static final ILog log = LogFactory.getLog(SyncControllerTask.class); @Autowired private KafkaControllerService kafkaControllerService; @Override - public TaskResult processSubTask(ClusterPhy clusterPhy, long triggerTimeUnitMs) { + public TaskResult processClusterTask(ClusterPhy clusterPhy, long triggerTimeUnitMs) { Result controllerResult = kafkaControllerService.getControllerFromKafka(clusterPhy); if (controllerResult.failed()) { return new TaskResult(TaskResult.FAIL_CODE, controllerResult.getMessage()); diff --git a/km-task/src/main/java/com/xiaojukeji/know/streaming/km/task/metadata/SyncKafkaAclTask.java b/km-task/src/main/java/com/xiaojukeji/know/streaming/km/task/metadata/SyncKafkaAclTask.java index 58ee3dd4..8b87d36e 100644 --- a/km-task/src/main/java/com/xiaojukeji/know/streaming/km/task/metadata/SyncKafkaAclTask.java +++ b/km-task/src/main/java/com/xiaojukeji/know/streaming/km/task/metadata/SyncKafkaAclTask.java @@ -11,7 +11,6 @@ import com.xiaojukeji.know.streaming.km.common.bean.po.KafkaAclPO; import com.xiaojukeji.know.streaming.km.common.converter.KafkaAclConverter; import com.xiaojukeji.know.streaming.km.core.service.acl.KafkaAclService; import com.xiaojukeji.know.streaming.km.core.service.acl.OpKafkaAclService; -import com.xiaojukeji.know.streaming.km.task.AbstractClusterPhyDispatchTask; import org.apache.kafka.common.acl.AclBinding; import org.springframework.beans.factory.annotation.Autowired; @@ -19,12 +18,12 @@ import java.util.List; import java.util.stream.Collectors; @Task(name = "SyncKafkaAclTask", - description = "KafkaAcl信息同步到DB,", + description = "KafkaAcl信息同步到DB", cron = "0 0/1 * * * ? *", autoRegister = true, consensual = ConsensualEnum.BROADCAST, timeout = 2 * 60) -public class SyncKafkaAclTask extends AbstractClusterPhyDispatchTask { +public class SyncKafkaAclTask extends AbstractAsyncMetadataDispatchTask { private static final ILog log = LogFactory.getLog(SyncKafkaAclTask.class); @Autowired @@ -34,7 +33,7 @@ public class SyncKafkaAclTask extends AbstractClusterPhyDispatchTask { private OpKafkaAclService opKafkaAclService; @Override - public TaskResult processSubTask(ClusterPhy clusterPhy, long triggerTimeUnitMs) { + public TaskResult processClusterTask(ClusterPhy clusterPhy, long triggerTimeUnitMs) { Result> aclBindingListResult = kafkaAclService.getAclFromKafka(clusterPhy.getId()); if (aclBindingListResult.failed()) { return TaskResult.FAIL; diff --git a/km-task/src/main/java/com/xiaojukeji/know/streaming/km/task/metadata/SyncKafkaGroupTask.java b/km-task/src/main/java/com/xiaojukeji/know/streaming/km/task/metadata/SyncKafkaGroupTask.java index 8861d085..65b64b96 100644 --- a/km-task/src/main/java/com/xiaojukeji/know/streaming/km/task/metadata/SyncKafkaGroupTask.java +++ b/km-task/src/main/java/com/xiaojukeji/know/streaming/km/task/metadata/SyncKafkaGroupTask.java @@ -12,7 +12,6 @@ import com.xiaojukeji.know.streaming.km.common.exception.AdminOperateException; import com.xiaojukeji.know.streaming.km.common.exception.NotExistException; import com.xiaojukeji.know.streaming.km.common.utils.ValidateUtils; import com.xiaojukeji.know.streaming.km.core.service.group.GroupService; -import com.xiaojukeji.know.streaming.km.task.AbstractClusterPhyDispatchTask; import org.apache.kafka.clients.admin.*; import org.apache.kafka.common.TopicPartition; import org.springframework.beans.factory.annotation.Autowired; @@ -22,19 +21,19 @@ import java.util.stream.Collectors; @Task(name = "SyncKafkaGroupTask", - description = "KafkaGroup信息同步到DB,", + description = "KafkaGroup信息同步到DB", cron = "0 0/1 * * * ? *", autoRegister = true, consensual = ConsensualEnum.BROADCAST, timeout = 2 * 60) -public class SyncKafkaGroupTask extends AbstractClusterPhyDispatchTask { +public class SyncKafkaGroupTask extends AbstractAsyncMetadataDispatchTask { private static final ILog log = LogFactory.getLog(SyncKafkaGroupTask.class); @Autowired private GroupService groupService; @Override - public TaskResult processSubTask(ClusterPhy clusterPhy, long triggerTimeUnitMs) throws Exception { + public TaskResult processClusterTask(ClusterPhy clusterPhy, long triggerTimeUnitMs) throws Exception { TaskResult tr = TaskResult.SUCCESS; List groupNameList = groupService.listGroupsFromKafka(clusterPhy.getId()); diff --git a/km-task/src/main/java/com/xiaojukeji/know/streaming/km/task/metadata/SyncKafkaUserTask.java b/km-task/src/main/java/com/xiaojukeji/know/streaming/km/task/metadata/SyncKafkaUserTask.java index e716d4ee..2e2fd63e 100644 --- a/km-task/src/main/java/com/xiaojukeji/know/streaming/km/task/metadata/SyncKafkaUserTask.java +++ b/km-task/src/main/java/com/xiaojukeji/know/streaming/km/task/metadata/SyncKafkaUserTask.java @@ -9,26 +9,25 @@ import com.xiaojukeji.know.streaming.km.common.bean.entity.cluster.ClusterPhy; import com.xiaojukeji.know.streaming.km.common.bean.entity.kafkauser.KafkaUser; import com.xiaojukeji.know.streaming.km.common.bean.entity.result.Result; import com.xiaojukeji.know.streaming.km.core.service.kafkauser.KafkaUserService; -import com.xiaojukeji.know.streaming.km.task.AbstractClusterPhyDispatchTask; import org.springframework.beans.factory.annotation.Autowired; import java.util.List; import java.util.stream.Collectors; @Task(name = "SyncKafkaUserTask", - description = "KafkaUser信息同步到DB,", + description = "KafkaUser信息同步到DB", cron = "0 0/1 * * * ? *", autoRegister = true, consensual = ConsensualEnum.BROADCAST, timeout = 2 * 60) -public class SyncKafkaUserTask extends AbstractClusterPhyDispatchTask { +public class SyncKafkaUserTask extends AbstractAsyncMetadataDispatchTask { private static final ILog log = LogFactory.getLog(SyncKafkaUserTask.class); @Autowired private KafkaUserService kafkaUserService; @Override - public TaskResult processSubTask(ClusterPhy clusterPhy, long triggerTimeUnitMs) { + public TaskResult processClusterTask(ClusterPhy clusterPhy, long triggerTimeUnitMs) { // 查询KafkaUser数据 Result> kafkaUserResult = kafkaUserService.getKafkaUserFromKafka(clusterPhy.getId()); if (kafkaUserResult.failed()) { diff --git a/km-task/src/main/java/com/xiaojukeji/know/streaming/km/task/metadata/SyncPartitionTask.java b/km-task/src/main/java/com/xiaojukeji/know/streaming/km/task/metadata/SyncPartitionTask.java index b1edc024..e4b07ec3 100644 --- a/km-task/src/main/java/com/xiaojukeji/know/streaming/km/task/metadata/SyncPartitionTask.java +++ b/km-task/src/main/java/com/xiaojukeji/know/streaming/km/task/metadata/SyncPartitionTask.java @@ -10,7 +10,6 @@ import com.xiaojukeji.know.streaming.km.common.bean.entity.partition.Partition; import com.xiaojukeji.know.streaming.km.common.bean.entity.result.Result; import com.xiaojukeji.know.streaming.km.common.bean.po.partition.PartitionPO; import com.xiaojukeji.know.streaming.km.core.service.partition.PartitionService; -import com.xiaojukeji.know.streaming.km.task.AbstractClusterPhyDispatchTask; import org.springframework.beans.factory.annotation.Autowired; import java.util.ArrayList; @@ -19,19 +18,19 @@ import java.util.List; import java.util.Map; @Task(name = "SyncPartitionTask", - description = "Partition信息同步到DB,", + description = "Partition信息同步到DB", cron = "0 0/1 * * * ? *", autoRegister = true, consensual = ConsensualEnum.BROADCAST, timeout = 2 * 60) -public class SyncPartitionTask extends AbstractClusterPhyDispatchTask { +public class SyncPartitionTask extends AbstractAsyncMetadataDispatchTask { private static final ILog log = LogFactory.getLog(SyncPartitionTask.class); @Autowired private PartitionService partitionService; @Override - public TaskResult processSubTask(ClusterPhy clusterPhy, long triggerTimeUnitMs) { + public TaskResult processClusterTask(ClusterPhy clusterPhy, long triggerTimeUnitMs) { Result>> partitionsResult = partitionService.listPartitionsFromKafka(clusterPhy); if (partitionsResult.failed()) { return new TaskResult(TaskResult.FAIL_CODE, partitionsResult.getMessage()); diff --git a/km-task/src/main/java/com/xiaojukeji/know/streaming/km/task/metadata/SyncTopicConfigTask.java b/km-task/src/main/java/com/xiaojukeji/know/streaming/km/task/metadata/SyncTopicConfigTask.java index 1940c6b8..69b71d06 100644 --- a/km-task/src/main/java/com/xiaojukeji/know/streaming/km/task/metadata/SyncTopicConfigTask.java +++ b/km-task/src/main/java/com/xiaojukeji/know/streaming/km/task/metadata/SyncTopicConfigTask.java @@ -12,7 +12,6 @@ import com.xiaojukeji.know.streaming.km.common.bean.entity.topic.TopicConfig; import com.xiaojukeji.know.streaming.km.common.utils.ConvertUtil; import com.xiaojukeji.know.streaming.km.core.service.topic.TopicConfigService; import com.xiaojukeji.know.streaming.km.core.service.topic.TopicService; -import com.xiaojukeji.know.streaming.km.task.AbstractClusterPhyDispatchTask; import org.springframework.beans.factory.annotation.Autowired; import java.util.ArrayList; @@ -27,12 +26,12 @@ import java.util.Map; * @date 22/02/25 */ @Task(name = "SyncTopicConfigTask", - description = "Topic保存时间配置同步DB,", + description = "Topic保存时间配置同步DB", cron = "0 0/1 * * * ? *", autoRegister = true, consensual = ConsensualEnum.BROADCAST, timeout = 2 * 60) -public class SyncTopicConfigTask extends AbstractClusterPhyDispatchTask { +public class SyncTopicConfigTask extends AbstractAsyncMetadataDispatchTask { protected static final ILog log = LogFactory.getLog(SyncTopicConfigTask.class); @Autowired @@ -42,7 +41,7 @@ public class SyncTopicConfigTask extends AbstractClusterPhyDispatchTask { private TopicConfigService kafkaConfigService; @Override - public TaskResult processSubTask(ClusterPhy clusterPhy, long triggerTimeUnitMs) { + public TaskResult processClusterTask(ClusterPhy clusterPhy, long triggerTimeUnitMs) { boolean success = true; List topicConfigList = new ArrayList<>(); diff --git a/km-task/src/main/java/com/xiaojukeji/know/streaming/km/task/metadata/SyncTopicTask.java b/km-task/src/main/java/com/xiaojukeji/know/streaming/km/task/metadata/SyncTopicTask.java index 2b851566..4ed42c64 100644 --- a/km-task/src/main/java/com/xiaojukeji/know/streaming/km/task/metadata/SyncTopicTask.java +++ b/km-task/src/main/java/com/xiaojukeji/know/streaming/km/task/metadata/SyncTopicTask.java @@ -9,25 +9,24 @@ import com.xiaojukeji.know.streaming.km.common.bean.entity.cluster.ClusterPhy; import com.xiaojukeji.know.streaming.km.common.bean.entity.result.Result; import com.xiaojukeji.know.streaming.km.common.bean.entity.topic.Topic; import com.xiaojukeji.know.streaming.km.core.service.topic.TopicService; -import com.xiaojukeji.know.streaming.km.task.AbstractClusterPhyDispatchTask; import org.springframework.beans.factory.annotation.Autowired; import java.util.List; @Task(name = "SyncTopicTask", - description = "Topic信息同步到DB,", + description = "Topic信息同步到DB", cron = "0 0/1 * * * ? *", autoRegister = true, consensual = ConsensualEnum.BROADCAST, timeout = 2 * 60) -public class SyncTopicTask extends AbstractClusterPhyDispatchTask { +public class SyncTopicTask extends AbstractAsyncMetadataDispatchTask { private static final ILog log = LogFactory.getLog(SyncTopicTask.class); @Autowired private TopicService topicService; @Override - public TaskResult processSubTask(ClusterPhy clusterPhy, long triggerTimeUnitMs) { + public TaskResult processClusterTask(ClusterPhy clusterPhy, long triggerTimeUnitMs) { Result> topicsResult = topicService.listTopicsFromKafka(clusterPhy); if (topicsResult.failed()) { return new TaskResult(TaskResult.FAIL_CODE, topicsResult.getMessage()); diff --git a/km-task/src/main/java/com/xiaojukeji/know/streaming/km/task/metrics/AbstractAsyncMetricsDispatchTask.java b/km-task/src/main/java/com/xiaojukeji/know/streaming/km/task/metrics/AbstractAsyncMetricsDispatchTask.java new file mode 100644 index 00000000..705c2301 --- /dev/null +++ b/km-task/src/main/java/com/xiaojukeji/know/streaming/km/task/metrics/AbstractAsyncMetricsDispatchTask.java @@ -0,0 +1,47 @@ +package com.xiaojukeji.know.streaming.km.task.metrics; + +import com.didiglobal.logi.job.common.TaskResult; +import com.didiglobal.logi.log.ILog; +import com.didiglobal.logi.log.LogFactory; +import com.xiaojukeji.know.streaming.km.common.bean.entity.cluster.ClusterPhy; +import com.xiaojukeji.know.streaming.km.task.AbstractClusterPhyDispatchTask; +import com.xiaojukeji.know.streaming.km.task.service.TaskThreadPoolService; +import org.springframework.beans.factory.annotation.Autowired; + +/** + * 指标采集相关任务 + */ +public abstract class AbstractAsyncMetricsDispatchTask extends AbstractClusterPhyDispatchTask { + private static final ILog log = LogFactory.getLog(AbstractAsyncMetricsDispatchTask.class); + + public abstract TaskResult processClusterTask(ClusterPhy clusterPhy, long triggerTimeUnitMs) throws Exception; + + @Autowired + private TaskThreadPoolService taskThreadPoolService; + + @Override + protected TaskResult processSubTask(ClusterPhy clusterPhy, long triggerTimeUnitMs) throws Exception { + return this.asyncProcessSubTask(clusterPhy, triggerTimeUnitMs); + } + + public TaskResult asyncProcessSubTask(ClusterPhy clusterPhy, long triggerTimeUnitMs) { + taskThreadPoolService.submitMetricsTask( + String.format("taskName=%s||clusterPhyId=%d", this.taskName, clusterPhy.getId()), + this.timeoutUnitSec.intValue() * 1000, + () -> { + try { + TaskResult tr = this.processClusterTask(clusterPhy, triggerTimeUnitMs); + if (TaskResult.SUCCESS_CODE != tr.getCode()) { + log.error("class=AbstractAsyncMetricsDispatchTask||taskName={}||clusterPhyId={}||taskResult={}||msg=failed", this.taskName, clusterPhy.getId(), tr); + } else { + log.debug("class=AbstractAsyncMetricsDispatchTask||taskName={}||clusterPhyId={}||msg=success", this.taskName, clusterPhy.getId()); + } + } catch (Exception e) { + log.error("class=AbstractAsyncMetricsDispatchTask||taskName={}||clusterPhyId={}||errMsg=exception", this.taskName, clusterPhy.getId(), e); + } + } + ); + + return TaskResult.SUCCESS; + } +} diff --git a/km-task/src/main/java/com/xiaojukeji/know/streaming/km/task/metrics/BrokerMetricCollectorTask.java b/km-task/src/main/java/com/xiaojukeji/know/streaming/km/task/metrics/BrokerMetricCollectorTask.java index 486b66b0..16334246 100644 --- a/km-task/src/main/java/com/xiaojukeji/know/streaming/km/task/metrics/BrokerMetricCollectorTask.java +++ b/km-task/src/main/java/com/xiaojukeji/know/streaming/km/task/metrics/BrokerMetricCollectorTask.java @@ -7,35 +7,26 @@ import com.didiglobal.logi.log.ILog; import com.didiglobal.logi.log.LogFactory; import com.xiaojukeji.know.streaming.km.collector.metric.BrokerMetricCollector; import com.xiaojukeji.know.streaming.km.common.bean.entity.cluster.ClusterPhy; -import com.xiaojukeji.know.streaming.km.task.AbstractClusterPhyDispatchTask; -import com.xiaojukeji.know.streaming.km.task.service.TaskThreadPoolService; import org.springframework.beans.factory.annotation.Autowired; /** * @author didi */ @Task(name = "BrokerMetricCollectorTask", - description = "Broker指标采集任务,", + description = "Broker指标采集任务", cron = "0 0/1 * * * ? *", autoRegister = true, consensual = ConsensualEnum.BROADCAST, timeout = 2 * 60) -public class BrokerMetricCollectorTask extends AbstractClusterPhyDispatchTask { +public class BrokerMetricCollectorTask extends AbstractAsyncMetricsDispatchTask { private static final ILog log = LogFactory.getLog(BrokerMetricCollectorTask.class); @Autowired private BrokerMetricCollector brokerMetricCollector; - @Autowired - private TaskThreadPoolService taskThreadPoolService; - @Override - public TaskResult processSubTask(ClusterPhy clusterPhy, long triggerTimeUnitMs) throws Exception { - taskThreadPoolService.submitHeavenTask( - String.format("TaskName=%s clusterPhyId=%d", this.taskName, clusterPhy.getId()), - 100000, - () -> brokerMetricCollector.collectMetrics(clusterPhy) - ); + public TaskResult processClusterTask(ClusterPhy clusterPhy, long triggerTimeUnitMs) throws Exception { + brokerMetricCollector.collectMetrics(clusterPhy); return TaskResult.SUCCESS; } diff --git a/km-task/src/main/java/com/xiaojukeji/know/streaming/km/task/metrics/ClusterMetricCollectorTask.java b/km-task/src/main/java/com/xiaojukeji/know/streaming/km/task/metrics/ClusterMetricCollectorTask.java index 2b6a3b3b..4abf0372 100644 --- a/km-task/src/main/java/com/xiaojukeji/know/streaming/km/task/metrics/ClusterMetricCollectorTask.java +++ b/km-task/src/main/java/com/xiaojukeji/know/streaming/km/task/metrics/ClusterMetricCollectorTask.java @@ -7,35 +7,26 @@ import com.didiglobal.logi.log.ILog; import com.didiglobal.logi.log.LogFactory; import com.xiaojukeji.know.streaming.km.collector.metric.ClusterMetricCollector; import com.xiaojukeji.know.streaming.km.common.bean.entity.cluster.ClusterPhy; -import com.xiaojukeji.know.streaming.km.task.AbstractClusterPhyDispatchTask; -import com.xiaojukeji.know.streaming.km.task.service.TaskThreadPoolService; import org.springframework.beans.factory.annotation.Autowired; /** * @author didi */ @Task(name = "ClusterMetricCollectorTask", - description = "Cluster指标采集任务,", + description = "Cluster指标采集任务", cron = "0 0/1 * * * ? *", autoRegister = true, consensual = ConsensualEnum.BROADCAST, timeout = 2 * 60) -public class ClusterMetricCollectorTask extends AbstractClusterPhyDispatchTask { +public class ClusterMetricCollectorTask extends AbstractAsyncMetricsDispatchTask { private static final ILog log = LogFactory.getLog(ClusterMetricCollectorTask.class); @Autowired private ClusterMetricCollector clusterMetricCollector; - @Autowired - private TaskThreadPoolService taskThreadPoolService; - @Override - public TaskResult processSubTask(ClusterPhy clusterPhy, long triggerTimeUnitMs) throws Exception { - taskThreadPoolService.submitHeavenTask( - String.format("TaskName=%s clusterPhyId=%d", this.taskName, clusterPhy.getId()), - 100000, - () -> clusterMetricCollector.collectMetrics(clusterPhy) - ); + public TaskResult processClusterTask(ClusterPhy clusterPhy, long triggerTimeUnitMs) throws Exception { + clusterMetricCollector.collectMetrics(clusterPhy); return TaskResult.SUCCESS; } diff --git a/km-task/src/main/java/com/xiaojukeji/know/streaming/km/task/metrics/GroupMetricCollectorTask.java b/km-task/src/main/java/com/xiaojukeji/know/streaming/km/task/metrics/GroupMetricCollectorTask.java index 5ca97906..595930b8 100644 --- a/km-task/src/main/java/com/xiaojukeji/know/streaming/km/task/metrics/GroupMetricCollectorTask.java +++ b/km-task/src/main/java/com/xiaojukeji/know/streaming/km/task/metrics/GroupMetricCollectorTask.java @@ -7,35 +7,26 @@ import com.didiglobal.logi.log.ILog; import com.didiglobal.logi.log.LogFactory; import com.xiaojukeji.know.streaming.km.collector.metric.GroupMetricCollector; import com.xiaojukeji.know.streaming.km.common.bean.entity.cluster.ClusterPhy; -import com.xiaojukeji.know.streaming.km.task.AbstractClusterPhyDispatchTask; -import com.xiaojukeji.know.streaming.km.task.service.TaskThreadPoolService; import org.springframework.beans.factory.annotation.Autowired; /** * @author didi */ @Task(name = "GroupMetricCollectorTask", - description = "Group指标采集任务,", + description = "Group指标采集任务", cron = "0 0/1 * * * ? *", autoRegister = true, consensual = ConsensualEnum.BROADCAST, timeout = 2 * 60) -public class GroupMetricCollectorTask extends AbstractClusterPhyDispatchTask { +public class GroupMetricCollectorTask extends AbstractAsyncMetricsDispatchTask { private static final ILog log = LogFactory.getLog(GroupMetricCollectorTask.class); @Autowired private GroupMetricCollector groupMetricCollector; - @Autowired - private TaskThreadPoolService taskThreadPoolService; - @Override - public TaskResult processSubTask(ClusterPhy clusterPhy, long triggerTimeUnitMs) throws Exception { - taskThreadPoolService.submitHeavenTask( - String.format("TaskName=%s clusterPhyId=%d", this.taskName, clusterPhy.getId()), - 100000, - () -> groupMetricCollector.collectMetrics(clusterPhy) - ); + public TaskResult processClusterTask(ClusterPhy clusterPhy, long triggerTimeUnitMs) throws Exception { + groupMetricCollector.collectMetrics(clusterPhy); return TaskResult.SUCCESS; } diff --git a/km-task/src/main/java/com/xiaojukeji/know/streaming/km/task/metrics/PartitionMetricCollectorTask.java b/km-task/src/main/java/com/xiaojukeji/know/streaming/km/task/metrics/PartitionMetricCollectorTask.java index 47b2fc60..21f9133b 100644 --- a/km-task/src/main/java/com/xiaojukeji/know/streaming/km/task/metrics/PartitionMetricCollectorTask.java +++ b/km-task/src/main/java/com/xiaojukeji/know/streaming/km/task/metrics/PartitionMetricCollectorTask.java @@ -5,34 +5,25 @@ import com.didiglobal.logi.job.common.TaskResult; import com.didiglobal.logi.job.core.consensual.ConsensualEnum; import com.xiaojukeji.know.streaming.km.collector.metric.PartitionMetricCollector; import com.xiaojukeji.know.streaming.km.common.bean.entity.cluster.ClusterPhy; -import com.xiaojukeji.know.streaming.km.task.AbstractClusterPhyDispatchTask; -import com.xiaojukeji.know.streaming.km.task.service.TaskThreadPoolService; import org.springframework.beans.factory.annotation.Autowired; /** * @author didi */ @Task(name = "PartitionMetricCollectorTask", - description = "Partition指标采集任务,", + description = "Partition指标采集任务", cron = "0 0/1 * * * ? *", autoRegister = true, consensual = ConsensualEnum.BROADCAST, timeout = 2 * 60) -public class PartitionMetricCollectorTask extends AbstractClusterPhyDispatchTask { +public class PartitionMetricCollectorTask extends AbstractAsyncMetricsDispatchTask { @Autowired private PartitionMetricCollector partitionMetricCollector; - @Autowired - private TaskThreadPoolService taskThreadPoolService; - @Override - public TaskResult processSubTask(ClusterPhy clusterPhy, long triggerTimeUnitMs) throws Exception { - taskThreadPoolService.submitHeavenTask( - String.format("TaskName=%s clusterPhyId=%d", this.taskName, clusterPhy.getId()), - 100000, - () -> partitionMetricCollector.collectMetrics(clusterPhy) - ); + public TaskResult processClusterTask(ClusterPhy clusterPhy, long triggerTimeUnitMs) throws Exception { + partitionMetricCollector.collectMetrics(clusterPhy); return TaskResult.SUCCESS; } diff --git a/km-task/src/main/java/com/xiaojukeji/know/streaming/km/task/metrics/ReplicaMetricCollectorTask.java b/km-task/src/main/java/com/xiaojukeji/know/streaming/km/task/metrics/ReplicaMetricCollectorTask.java index 5e7a738b..6b93e324 100644 --- a/km-task/src/main/java/com/xiaojukeji/know/streaming/km/task/metrics/ReplicaMetricCollectorTask.java +++ b/km-task/src/main/java/com/xiaojukeji/know/streaming/km/task/metrics/ReplicaMetricCollectorTask.java @@ -5,8 +5,6 @@ import com.didiglobal.logi.job.common.TaskResult; import com.didiglobal.logi.job.core.consensual.ConsensualEnum; import com.xiaojukeji.know.streaming.km.collector.metric.ReplicaMetricCollector; import com.xiaojukeji.know.streaming.km.common.bean.entity.cluster.ClusterPhy; -import com.xiaojukeji.know.streaming.km.task.AbstractClusterPhyDispatchTask; -import com.xiaojukeji.know.streaming.km.task.service.TaskThreadPoolService; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Autowired; @@ -15,26 +13,19 @@ import org.springframework.beans.factory.annotation.Autowired; */ @Slf4j @Task(name = "ReplicaMetricCollectorTask", - description = "Replica指标采集任务,", + description = "Replica指标采集任务", cron = "0 0/1 * * * ? *", autoRegister = true, consensual = ConsensualEnum.BROADCAST, timeout = 2 * 60) -public class ReplicaMetricCollectorTask extends AbstractClusterPhyDispatchTask { +public class ReplicaMetricCollectorTask extends AbstractAsyncMetricsDispatchTask { @Autowired private ReplicaMetricCollector replicaMetricCollector; - @Autowired - private TaskThreadPoolService taskThreadPoolService; - @Override - protected TaskResult processSubTask(ClusterPhy clusterPhy, long triggerTimeUnitMs) throws Exception { - taskThreadPoolService.submitHeavenTask( - String.format("TaskName=%s clusterPhyId=%d", this.taskName, clusterPhy.getId()), - 100000, - () -> replicaMetricCollector.collectMetrics(clusterPhy) - ); + public TaskResult processClusterTask(ClusterPhy clusterPhy, long triggerTimeUnitMs) throws Exception { + replicaMetricCollector.collectMetrics(clusterPhy); return TaskResult.SUCCESS; } diff --git a/km-task/src/main/java/com/xiaojukeji/know/streaming/km/task/metrics/TopicMetricCollectorTask.java b/km-task/src/main/java/com/xiaojukeji/know/streaming/km/task/metrics/TopicMetricCollectorTask.java index 9c3b7863..9f8a5de7 100644 --- a/km-task/src/main/java/com/xiaojukeji/know/streaming/km/task/metrics/TopicMetricCollectorTask.java +++ b/km-task/src/main/java/com/xiaojukeji/know/streaming/km/task/metrics/TopicMetricCollectorTask.java @@ -7,8 +7,6 @@ import com.didiglobal.logi.log.ILog; import com.didiglobal.logi.log.LogFactory; import com.xiaojukeji.know.streaming.km.collector.metric.TopicMetricCollector; import com.xiaojukeji.know.streaming.km.common.bean.entity.cluster.ClusterPhy; -import com.xiaojukeji.know.streaming.km.task.AbstractClusterPhyDispatchTask; -import com.xiaojukeji.know.streaming.km.task.service.TaskThreadPoolService; import org.springframework.beans.factory.annotation.Autowired; /** @@ -20,22 +18,15 @@ import org.springframework.beans.factory.annotation.Autowired; autoRegister = true, consensual = ConsensualEnum.BROADCAST, timeout = 2 * 60) -public class TopicMetricCollectorTask extends AbstractClusterPhyDispatchTask { +public class TopicMetricCollectorTask extends AbstractAsyncMetricsDispatchTask { private static final ILog log = LogFactory.getLog(TopicMetricCollectorTask.class); @Autowired private TopicMetricCollector topicMetricCollector; - @Autowired - private TaskThreadPoolService taskThreadPoolService; - @Override - public TaskResult processSubTask(ClusterPhy clusterPhy, long triggerTimeUnitMs) throws Exception { - taskThreadPoolService.submitHeavenTask( - String.format("TaskName=%s clusterPhyId=%d", this.taskName, clusterPhy.getId()), - 100000, - () -> topicMetricCollector.collectMetrics(clusterPhy) - ); + public TaskResult processClusterTask(ClusterPhy clusterPhy, long triggerTimeUnitMs) throws Exception { + topicMetricCollector.collectMetrics(clusterPhy); return TaskResult.SUCCESS; } 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 884a572d..f4bb6826 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 @@ -16,23 +16,72 @@ import javax.annotation.PostConstruct; @NoArgsConstructor public class TaskThreadPoolService { /** - * 较重任务，比如指标采集 + * metrics任务，比如指标采集 */ - private FutureNoWaitUtil heavenTaskThreadPool; + private FutureNoWaitUtil metricsTaskThreadPool; + + @Value(value = "${thread-pool.task.metrics.thread-num:18}") + private Integer metricsTaskThreadNum; + + @Value(value = "${thread-pool.task.metrics.queue-size:180}") + private Integer metricsTaskQueueSize; - @Value(value = "${thread-pool.task.heaven.thread-num:12}") - private Integer heavenTaskThreadNum; + /** + * metadata任务 + */ + private FutureNoWaitUtil metadataTaskThreadPool; - @Value(value = "${thread-pool.task.heaven.queue-size:1000}") - private Integer heavenTaskQueueSize; + @Value(value = "${thread-pool.task.metadata.thread-num:27}") + private Integer metadataTaskThreadNum; + + @Value(value = "${thread-pool.task.metadata.queue-size:270}") + private Integer metadataTaskQueueSize; + + /** + * common任务 + */ + private FutureNoWaitUtil commonTaskThreadPool; + + @Value(value = "${thread-pool.task.common.thread-num:15}") + private Integer commonTaskThreadNum; + + @Value(value = "${thread-pool.task.common.queue-size:150}") + private Integer commonTaskQueueSize; @PostConstruct private void init() { - heavenTaskThreadPool = FutureNoWaitUtil.init("heavenTaskThreadPool", heavenTaskThreadNum, heavenTaskThreadNum, heavenTaskQueueSize); + metricsTaskThreadPool = FutureNoWaitUtil.init( + "metricsTaskThreadPool", + metricsTaskThreadNum, + metricsTaskThreadNum, + metricsTaskQueueSize + ); + + metadataTaskThreadPool = FutureNoWaitUtil.init( + "metadataTaskThreadPool", + metadataTaskThreadNum, + metadataTaskThreadNum, + metadataTaskQueueSize + ); + + commonTaskThreadPool = FutureNoWaitUtil.init( + "commonTaskThreadPool", + commonTaskThreadNum, + commonTaskThreadNum, + commonTaskQueueSize + ); } - public void submitHeavenTask(String taskName, Integer timeoutUnisMs, Runnable runnable) { - heavenTaskThreadPool.runnableTask(taskName, timeoutUnisMs, runnable); + public void submitMetricsTask(String taskName, Integer timeoutUnisMs, Runnable runnable) { + metricsTaskThreadPool.runnableTask(taskName, timeoutUnisMs, runnable); + } + + public void submitMetadataTask(String taskName, Integer timeoutUnisMs, Runnable runnable) { + metadataTaskThreadPool.runnableTask(taskName, timeoutUnisMs, runnable); + } + + public void submitCommonTask(String taskName, Integer timeoutUnisMs, Runnable runnable) { + commonTaskThreadPool.runnableTask(taskName, timeoutUnisMs, runnable); } }