rocketmq-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From wlliqip...@apache.org
Subject [rocketmq-exporter] 39/43: add all metrics
Date Sun, 08 Mar 2020 13:16:59 GMT
This is an automated email from the ASF dual-hosted git repository.

wlliqipeng pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/rocketmq-exporter.git

commit f0e1125a8ca54e65edf299aab66c38e18564355b
Author: liwei5 <liwei5@vipkid.com.cn>
AuthorDate: Thu Dec 12 00:17:32 2019 +0800

    add all metrics
---
 pom.xml                                            |   10 +-
 .../exporter/aspect/admin/MQAdminAspect.java       |   71 --
 .../admin/annotation/MultiMQAdminCmdMethod.java    |   30 -
 .../exporter/collector/RMQMetricsCollector.java    | 1178 ++++++++++++++++++--
 .../rocketmq/exporter/config/RMQConfigure.java     |    4 +-
 .../rocketmq/exporter/config/ScheduleConfig.java   |   26 +
 .../exporter/controller/RMQMetricsController.java  |    9 +-
 .../exporter/exception/ServiceException.java       |   31 -
 .../exporter/model/BrokerRuntimeStats.java         |  609 ++++++++++
 .../exporter/model/metrics/BrokerMetric.java       |   43 +-
 .../model/metrics/ConsumerCountMetric.java         |   58 +
 .../exporter/model/metrics/ConsumerMetric.java     |   45 +-
 .../model/metrics/ConsumerQueueMetric.java         |   93 --
 .../model/metrics/ConsumerTopicDiffMetric.java     |   61 +
 .../model/metrics/DLQTopicOffsetMetric.java        |   72 ++
 .../exporter/model/metrics/ProducerMetric.java     |   61 +-
 .../exporter/model/metrics/TopicPutNumMetric.java  |   83 ++
 .../metrics/brokerruntime/BrokerRuntimeMetric.java |   91 ++
 .../exporter/service/AbstractCommonService.java    |   50 -
 .../exporter/service/RMQMetricsService.java        |    6 +-
 .../exporter/service/client/MQAdminExtImpl.java    |  279 ++---
 .../exporter/service/client/MQAdminInstance.java   |  118 +-
 .../service/impl/RMQMetricsServiceImpl.java        |   16 +-
 .../rocketmq/exporter/task/MetricsCollectTask.java |  603 ++++++----
 .../apache/rocketmq/exporter/util/JsonUtil.java    |   39 +-
 src/main/resources/application.properties          |   14 -
 src/main/resources/application.yml                 |   32 +
 src/main/resources/logback.xml                     |   55 +-
 28 files changed, 2849 insertions(+), 938 deletions(-)

diff --git a/pom.xml b/pom.xml
index cda549d..8a84c73 100644
--- a/pom.xml
+++ b/pom.xml
@@ -130,7 +130,15 @@
 					</resources>
 				</configuration>
 			</plugin>
-		</plugins>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-compiler-plugin</artifactId>
+                <configuration>
+                    <source>8</source>
+                    <target>8</target>
+                </configuration>
+            </plugin>
+        </plugins>
 
 	</build>
 
diff --git a/src/main/java/org/apache/rocketmq/exporter/aspect/admin/MQAdminAspect.java b/src/main/java/org/apache/rocketmq/exporter/aspect/admin/MQAdminAspect.java
deleted file mode 100644
index 6b5435f..0000000
--- a/src/main/java/org/apache/rocketmq/exporter/aspect/admin/MQAdminAspect.java
+++ /dev/null
@@ -1,71 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- *  Unless required by applicable law or agreed to in writing, software
- *  distributed under the License is distributed on an "AS IS" BASIS,
- *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- *  See the License for the specific language governing permissions and
- *  limitations under the License.
- */
-package org.apache.rocketmq.exporter.aspect.admin;
-
-import java.lang.reflect.Method;
-import org.apache.rocketmq.exporter.aspect.admin.annotation.MultiMQAdminCmdMethod;
-import org.apache.rocketmq.exporter.service.client.MQAdminInstance;
-import org.aspectj.lang.ProceedingJoinPoint;
-import org.aspectj.lang.annotation.Around;
-import org.aspectj.lang.annotation.Aspect;
-import org.aspectj.lang.annotation.Pointcut;
-import org.aspectj.lang.reflect.MethodSignature;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.springframework.stereotype.Service;
-
-@Aspect
-@Service
-public class MQAdminAspect {
-    private Logger logger = LoggerFactory.getLogger(MQAdminAspect.class);
-
-    public MQAdminAspect() {
-    }
-
-    @Pointcut("execution(* org.apache.rocketmq.exporter.service.client.MQAdminExtImpl..*(..))")
-    public void mQAdminMethodPointCut() {
-
-    }
-
-    @Pointcut("@annotation(org.apache.rocketmq.exporter.aspect.admin.annotation.MultiMQAdminCmdMethod)")
-    public void multiMQAdminMethodPointCut() {
-
-    }
-
-    @Around(value = "mQAdminMethodPointCut() || multiMQAdminMethodPointCut()")
-    public Object aroundMQAdminMethod(ProceedingJoinPoint joinPoint) throws Throwable {
-        long start = System.currentTimeMillis();
-        Object obj;
-        try {
-            MethodSignature signature = (MethodSignature)joinPoint.getSignature();
-            Method method = signature.getMethod();
-            MultiMQAdminCmdMethod multiMQAdminCmdMethod = method.getAnnotation(MultiMQAdminCmdMethod.class);
-            if (multiMQAdminCmdMethod != null && multiMQAdminCmdMethod.timeoutMillis() > 0) {
-                MQAdminInstance.initMQAdminInstance(multiMQAdminCmdMethod.timeoutMillis());
-            }
-            else {
-                MQAdminInstance.initMQAdminInstance(0);
-            }
-            obj = joinPoint.proceed();
-        }
-        finally {
-            MQAdminInstance.destroyMQAdminInstance();
-            logger.debug("op=look method={} cost={}", joinPoint.getSignature().getName(), System.currentTimeMillis() - start);
-        }
-        return obj;
-    }
-}
diff --git a/src/main/java/org/apache/rocketmq/exporter/aspect/admin/annotation/MultiMQAdminCmdMethod.java b/src/main/java/org/apache/rocketmq/exporter/aspect/admin/annotation/MultiMQAdminCmdMethod.java
deleted file mode 100644
index 7953fb3..0000000
--- a/src/main/java/org/apache/rocketmq/exporter/aspect/admin/annotation/MultiMQAdminCmdMethod.java
+++ /dev/null
@@ -1,30 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- *  Unless required by applicable law or agreed to in writing, software
- *  distributed under the License is distributed on an "AS IS" BASIS,
- *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- *  See the License for the specific language governing permissions and
- *  limitations under the License.
- */
-package org.apache.rocketmq.exporter.aspect.admin.annotation;
-
-import java.lang.annotation.Documented;
-import java.lang.annotation.ElementType;
-import java.lang.annotation.Retention;
-import java.lang.annotation.RetentionPolicy;
-import java.lang.annotation.Target;
-
-@Target({ElementType.METHOD})
-@Retention(RetentionPolicy.RUNTIME)
-@Documented
-public @interface MultiMQAdminCmdMethod {
-    long timeoutMillis() default 0;
-}
diff --git a/src/main/java/org/apache/rocketmq/exporter/collector/RMQMetricsCollector.java b/src/main/java/org/apache/rocketmq/exporter/collector/RMQMetricsCollector.java
index 2854712..253e9f9 100644
--- a/src/main/java/org/apache/rocketmq/exporter/collector/RMQMetricsCollector.java
+++ b/src/main/java/org/apache/rocketmq/exporter/collector/RMQMetricsCollector.java
@@ -17,12 +17,17 @@
 package org.apache.rocketmq.exporter.collector;
 
 import io.prometheus.client.Collector;
-import io.prometheus.client.CounterMetricFamily;
 import io.prometheus.client.GaugeMetricFamily;
+import org.apache.rocketmq.common.MixAll;
+import org.apache.rocketmq.exporter.model.BrokerRuntimeStats;
 import org.apache.rocketmq.exporter.model.metrics.BrokerMetric;
+import org.apache.rocketmq.exporter.model.metrics.ConsumerCountMetric;
 import org.apache.rocketmq.exporter.model.metrics.ConsumerMetric;
-import org.apache.rocketmq.exporter.model.metrics.ConsumerQueueMetric;
+import org.apache.rocketmq.exporter.model.metrics.ConsumerTopicDiffMetric;
+import org.apache.rocketmq.exporter.model.metrics.DLQTopicOffsetMetric;
 import org.apache.rocketmq.exporter.model.metrics.ProducerMetric;
+import org.apache.rocketmq.exporter.model.metrics.TopicPutNumMetric;
+import org.apache.rocketmq.exporter.model.metrics.brokerruntime.BrokerRuntimeMetric;
 
 import java.util.ArrayList;
 import java.util.Arrays;
@@ -31,160 +36,1159 @@ import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 
 public class RMQMetricsCollector extends Collector {
+    //max offset of normal consume queue
+    private ConcurrentHashMap<ProducerMetric, Double> topicOffset = new ConcurrentHashMap<>();
+    //max offset of retry topic consume queue
+    private ConcurrentHashMap<ProducerMetric, Double> topicRetryOffset = new ConcurrentHashMap<>();
+    //max offset of dlq consume queue
+    private ConcurrentHashMap<DLQTopicOffsetMetric, Double> topicDLQOffset = new ConcurrentHashMap<>();
 
-    private ConcurrentHashMap<ProducerMetric, Double>   topicPutNums            = new ConcurrentHashMap<>();
-    private ConcurrentHashMap<ProducerMetric, Double>   topicPutSize            = new ConcurrentHashMap<>();
+    //total put numbers for topics
+    private ConcurrentHashMap<TopicPutNumMetric, Double> topicPutNums = new ConcurrentHashMap<>();
+    //total get numbers for topics
+    private ConcurrentHashMap<TopicPutNumMetric, Double> topicPutSize = new ConcurrentHashMap<>();
 
-    private ConcurrentHashMap<ProducerMetric, Double>   topicOffset              = new ConcurrentHashMap<>();
+    //diff for consumer group
+    private ConcurrentHashMap<ConsumerTopicDiffMetric, Long> consumerDiff = new ConcurrentHashMap<>();
+    //retry diff for consumer group
+    private ConcurrentHashMap<ConsumerTopicDiffMetric, Long> consumerRetryDiff = new ConcurrentHashMap<>();
+    //死信堆积 todo 检查是否存在这个数据 应该不存在
+    private ConcurrentHashMap<ConsumerTopicDiffMetric, Long> consumerDLQDiff = new ConcurrentHashMap<>();
+    //consumer count
+    private ConcurrentHashMap<ConsumerCountMetric, Integer> consumerCounts = new ConcurrentHashMap<>();
 
-    private ConcurrentHashMap<BrokerMetric, Double>     brokerPutNums           = new ConcurrentHashMap<>();
-    private ConcurrentHashMap<BrokerMetric, Double>     brokerGetNums           = new ConcurrentHashMap<>();
+    //broker offset for consumer-topic
+    private ConcurrentHashMap<ConsumerMetric, Long> groupBrokerTotalOffset = new ConcurrentHashMap<>();
+    //consumer offset for consumer-topic
+    private ConcurrentHashMap<ConsumerMetric, Long> groupConsumeTotalOffset = new ConcurrentHashMap<>();
+    //consume tps
+    private ConcurrentHashMap<ConsumerMetric, Double> groupConsumeTPS = new ConcurrentHashMap<>();
+    //consumed message count for consumer-topic
+    private ConcurrentHashMap<ConsumerMetric, Double> groupGetNums = new ConcurrentHashMap<>();
+    //consumed message size(byte) for consumer-topic
+    private ConcurrentHashMap<ConsumerMetric, Double> groupGetSize = new ConcurrentHashMap<>();
+    //re-consumed message count for consumer-topic
+    private ConcurrentHashMap<ConsumerMetric, Double> sendBackNums = new ConcurrentHashMap<>();
 
-    private ConcurrentHashMap<ConsumerMetric, Double>   groupGetNums            = new ConcurrentHashMap<>();
-    private ConcurrentHashMap<ConsumerMetric, Double>   groupGetSize            = new ConcurrentHashMap<>();
+    //total put message count for one broker
+    private ConcurrentHashMap<BrokerMetric, Double> brokerPutNums = new ConcurrentHashMap<>();
+    //total get message count for one broker
+    private ConcurrentHashMap<BrokerMetric, Double> brokerGetNums = new ConcurrentHashMap<>();
 
-    private ConcurrentHashMap<ConsumerMetric, Double>       sendBackNums        = new ConcurrentHashMap<>();
-    private ConcurrentHashMap<ConsumerMetric, Double>       groupOffset         = new ConcurrentHashMap<>();
+    private ConcurrentHashMap<BrokerRuntimeMetric, Long> brokerRuntimeMsgPutTotalTodayNow = new ConcurrentHashMap<>();
+    private ConcurrentHashMap<BrokerRuntimeMetric, Long> brokerRuntimeMsgGetTotalTodayNow = new ConcurrentHashMap<>();
 
-    private ConcurrentHashMap<ConsumerQueueMetric, Double>  groupGetLatency     = new ConcurrentHashMap<>();
+    private ConcurrentHashMap<BrokerRuntimeMetric, Long> brokerRuntimeMsgGetTotalYesterdayMorning = new ConcurrentHashMap<>();
+    private ConcurrentHashMap<BrokerRuntimeMetric, Long> brokerRuntimeMsgPutTotalYesterdayMorning = new ConcurrentHashMap<>();
+    private ConcurrentHashMap<BrokerRuntimeMetric, Long> brokerRuntimeMsgGetTotalTodayMorning = new ConcurrentHashMap<>();
+    private ConcurrentHashMap<BrokerRuntimeMetric, Long> brokerRuntimeMsgPutTotalTodayMorning = new ConcurrentHashMap<>();
 
-    private ConcurrentHashMap<ConsumerMetric, Double>  groupGetLatencyByStoreTime     = new ConcurrentHashMap<>();
+    private ConcurrentHashMap<BrokerRuntimeMetric, Long> brokerRuntimeDispatchBehindBytes = new ConcurrentHashMap<>();
+    private ConcurrentHashMap<BrokerRuntimeMetric, Long> brokerRuntimePutMessageSizeTotal = new ConcurrentHashMap<>();
+
+    private ConcurrentHashMap<BrokerRuntimeMetric, Double> brokerRuntimePutMessageAverageSize = new ConcurrentHashMap<>();
+    private ConcurrentHashMap<BrokerRuntimeMetric, Long> brokerRuntimeQueryThreadPoolQueueCapacity = new ConcurrentHashMap<>();
+    private ConcurrentHashMap<BrokerRuntimeMetric, Long> brokerRuntimeRemainTransientStoreBufferNumbs = new ConcurrentHashMap<>();
+    private ConcurrentHashMap<BrokerRuntimeMetric, Long> brokerRuntimeEarliestMessageTimeStamp = new ConcurrentHashMap<>();
+    private ConcurrentHashMap<BrokerRuntimeMetric, Long> brokerRuntimePutMessageEntireTimeMax = new ConcurrentHashMap<>();
+    private ConcurrentHashMap<BrokerRuntimeMetric, Long> brokerRuntimeStartAcceptSendRequestTimeStamp = new ConcurrentHashMap<>();
+    private ConcurrentHashMap<BrokerRuntimeMetric, Long> brokerRuntimeSendThreadPoolQueueSize = new ConcurrentHashMap<>();
+    private ConcurrentHashMap<BrokerRuntimeMetric, Long> brokerRuntimePutMessageTimesTotal = new ConcurrentHashMap<>();
+    private ConcurrentHashMap<BrokerRuntimeMetric, Long> brokerRuntimeGetMessageEntireTimeMax = new ConcurrentHashMap<>();
+    private ConcurrentHashMap<BrokerRuntimeMetric, Long> brokerRuntimePageCacheLockTimeMills = new ConcurrentHashMap<>();
+    private ConcurrentHashMap<BrokerRuntimeMetric, Double> brokerRuntimeCommitLogDiskRatio = new ConcurrentHashMap<>();
+    private ConcurrentHashMap<BrokerRuntimeMetric, Double> brokerRuntimeConsumeQueueDiskRatio = new ConcurrentHashMap<>();
+
+    private ConcurrentHashMap<BrokerRuntimeMetric, Double> brokerRuntimeGetFoundTps600 = new ConcurrentHashMap<>();
+    private ConcurrentHashMap<BrokerRuntimeMetric, Double> brokerRuntimeGetFoundTps60 = new ConcurrentHashMap<>();
+    private ConcurrentHashMap<BrokerRuntimeMetric, Double> brokerRuntimeGetFoundTps10 = new ConcurrentHashMap<>();
+    private ConcurrentHashMap<BrokerRuntimeMetric, Double> brokerRuntimeGetTotalTps600 = new ConcurrentHashMap<>();
+    private ConcurrentHashMap<BrokerRuntimeMetric, Double> brokerRuntimeGetTotalTps60 = new ConcurrentHashMap<>();
+    private ConcurrentHashMap<BrokerRuntimeMetric, Double> brokerRuntimeGetTotalTps10 = new ConcurrentHashMap<>();
+    private ConcurrentHashMap<BrokerRuntimeMetric, Double> brokerRuntimeGetTransferedTps600 = new ConcurrentHashMap<>();
+    private ConcurrentHashMap<BrokerRuntimeMetric, Double> brokerRuntimeGetTransferedTps60 = new ConcurrentHashMap<>();
+    private ConcurrentHashMap<BrokerRuntimeMetric, Double> brokerRuntimeGetTransferedTps10 = new ConcurrentHashMap<>();
+    private ConcurrentHashMap<BrokerRuntimeMetric, Double> brokerRuntimeGetMissTps600 = new ConcurrentHashMap<>();
+    private ConcurrentHashMap<BrokerRuntimeMetric, Double> brokerRuntimeGetMissTps60 = new ConcurrentHashMap<>();
+    private ConcurrentHashMap<BrokerRuntimeMetric, Double> brokerRuntimeGetMissTps10 = new ConcurrentHashMap<>();
+    private ConcurrentHashMap<BrokerRuntimeMetric, Double> brokerRuntimePutTps600 = new ConcurrentHashMap<>();
+    private ConcurrentHashMap<BrokerRuntimeMetric, Double> brokerRuntimePutTps60 = new ConcurrentHashMap<>();
+    private ConcurrentHashMap<BrokerRuntimeMetric, Double> brokerRuntimePutTps10 = new ConcurrentHashMap<>();
+
+    private ConcurrentHashMap<BrokerRuntimeMetric, Long> brokerRuntimeDispatchMaxBuffer = new ConcurrentHashMap<>();
+
+    private ConcurrentHashMap<BrokerRuntimeMetric, Integer> brokerRuntimePutMessageDistributeTimeMap10toMore = new ConcurrentHashMap<>();
+    private ConcurrentHashMap<BrokerRuntimeMetric, Integer> brokerRuntimePutMessageDistributeTimeMap5to10s = new ConcurrentHashMap<>();
+    private ConcurrentHashMap<BrokerRuntimeMetric, Integer> brokerRuntimePutMessageDistributeTimeMap4to5s = new ConcurrentHashMap<>();
+    private ConcurrentHashMap<BrokerRuntimeMetric, Integer> brokerRuntimePutMessageDistributeTimeMap3to4s = new ConcurrentHashMap<>();
+    private ConcurrentHashMap<BrokerRuntimeMetric, Integer> brokerRuntimePutMessageDistributeTimeMap2to3s = new ConcurrentHashMap<>();
+    private ConcurrentHashMap<BrokerRuntimeMetric, Integer> brokerRuntimePutMessageDistributeTimeMap1to2s = new ConcurrentHashMap<>();
+    private ConcurrentHashMap<BrokerRuntimeMetric, Integer> brokerRuntimePutMessageDistributeTimeMap500to1s = new ConcurrentHashMap<>();
+    private ConcurrentHashMap<BrokerRuntimeMetric, Integer> brokerRuntimePutMessageDistributeTimeMap200to500ms = new ConcurrentHashMap<>();
+    private ConcurrentHashMap<BrokerRuntimeMetric, Integer> brokerRuntimePutMessageDistributeTimeMap100to200ms = new ConcurrentHashMap<>();
+    private ConcurrentHashMap<BrokerRuntimeMetric, Integer> brokerRuntimePutMessageDistributeTimeMap50to100ms = new ConcurrentHashMap<>();
+    private ConcurrentHashMap<BrokerRuntimeMetric, Integer> brokerRuntimePutMessageDistributeTimeMap10to50ms = new ConcurrentHashMap<>();
+    private ConcurrentHashMap<BrokerRuntimeMetric, Integer> brokerRuntimePutMessageDistributeTimeMap0to10ms = new ConcurrentHashMap<>();
+    private ConcurrentHashMap<BrokerRuntimeMetric, Integer> brokerRuntimePutMessageDistributeTimeMap0ms = new ConcurrentHashMap<>();
+
+    private ConcurrentHashMap<BrokerRuntimeMetric, Long> brokerRuntimePullThreadPoolQueueCapacity = new ConcurrentHashMap<>();
+    private ConcurrentHashMap<BrokerRuntimeMetric, Long> brokerRuntimeSendThreadPoolQueueCapacity = new ConcurrentHashMap<>();
+    private ConcurrentHashMap<BrokerRuntimeMetric, Long> brokerRuntimePullThreadPoolQueueSize = new ConcurrentHashMap<>();
+    private ConcurrentHashMap<BrokerRuntimeMetric, Long> brokerRuntimeQueryThreadPoolQueueSize = new ConcurrentHashMap<>();
+
+    private ConcurrentHashMap<BrokerRuntimeMetric, Long> brokerRuntimePullThreadPoolQueueHeadWaitTimeMills = new ConcurrentHashMap<>();
+    private ConcurrentHashMap<BrokerRuntimeMetric, Long> brokerRuntimeQueryThreadPoolQueueHeadWaitTimeMills = new ConcurrentHashMap<>();
+    private ConcurrentHashMap<BrokerRuntimeMetric, Long> brokerRuntimeSendThreadPoolQueueHeadWaitTimeMills = new ConcurrentHashMap<>();
+
+    private ConcurrentHashMap<BrokerRuntimeMetric, Double> brokerRuntimeCommitLogDirCapacityFree = new ConcurrentHashMap<>();
+    private ConcurrentHashMap<BrokerRuntimeMetric, Double> brokerRuntimeCommitLogDirCapacityTotal = new ConcurrentHashMap<>();
+
+    private ConcurrentHashMap<BrokerRuntimeMetric, Long> brokerRuntimeCommitLogMaxOffset = new ConcurrentHashMap<>();
+    private ConcurrentHashMap<BrokerRuntimeMetric, Long> brokerRuntimeCommitLogMinOffset = new ConcurrentHashMap<>();
+    private ConcurrentHashMap<BrokerRuntimeMetric, Double> brokerRuntimeRemainHowManyDataToFlush = new ConcurrentHashMap<>();
+
+    private static List<String> GROUP_DIFF_LABEL_NAMES = Arrays.asList("group", "topic", "countOfOnlineConsumers");
+
+    private static <T extends Number> void loadGroupDiffMetric(GaugeMetricFamily family, Map.Entry<ConsumerTopicDiffMetric, T> entry) {
+        family.addMetric(
+                Arrays.asList(
+                        entry.getKey().getGroup(),
+                        entry.getKey().getTopic(),
+                        entry.getKey().getCountOfOnlineConsumers()
+                ),
+                entry.getValue().doubleValue());
+    }
+
+    private static List<String> GROUP_COUNT_LABEL_NAMES = Arrays.asList("group", "caddr", "localaddr");
+
+    private void collectConsumerMetric(List<MetricFamilySamples> mfs) {
+        GaugeMetricFamily groupGetLatencyByConsumerDiff = new GaugeMetricFamily("rocketmq_group_diff", "GroupDiff", GROUP_DIFF_LABEL_NAMES);
+        for (Map.Entry<ConsumerTopicDiffMetric, Long> entry : consumerDiff.entrySet()) {
+            loadGroupDiffMetric(groupGetLatencyByConsumerDiff, entry);
+        }
+        mfs.add(groupGetLatencyByConsumerDiff);
+
+        GaugeMetricFamily groupGetLatencyByConsumerRetryDiff = new GaugeMetricFamily("rocketmq_group_retrydiff", "GroupRetryDiff", GROUP_DIFF_LABEL_NAMES);
+        for (Map.Entry<ConsumerTopicDiffMetric, Long> entry : consumerRetryDiff.entrySet()) {
+            loadGroupDiffMetric(groupGetLatencyByConsumerRetryDiff, entry);
+        }
+        mfs.add(groupGetLatencyByConsumerRetryDiff);
+
+        GaugeMetricFamily groupGetLatencyByConsumerDLQDiff = new GaugeMetricFamily("rocketmq_group_dlqdiff", "GroupDLQDiff", GROUP_DIFF_LABEL_NAMES);
+        for (Map.Entry<ConsumerTopicDiffMetric, Long> entry : consumerDLQDiff.entrySet()) {
+            loadGroupDiffMetric(groupGetLatencyByConsumerDLQDiff, entry);
+        }
+        mfs.add(groupGetLatencyByConsumerDLQDiff);
+
+        GaugeMetricFamily consumerCountsF = new GaugeMetricFamily("rocketmq_group_count", "GroupCount", GROUP_COUNT_LABEL_NAMES);
+        for (Map.Entry<ConsumerCountMetric, Integer> entry : consumerCounts.entrySet()) {
+            consumerCountsF.addMetric(
+                    Arrays.asList(
+                            entry.getKey().getGroup(),
+                            entry.getKey().getCaddr(),
+                            entry.getKey().getLocaladdr()
+                    ),
+                    entry.getValue().doubleValue());
+        }
+        mfs.add(consumerCountsF);
+
+    }
+
+
+    private static List<String> TOPIC_OFFSET_LABEL_NAMES = Arrays.asList(
+            "cluster", "brokerNames", "topic", "lastUpdateTimestamp"
+    );
+
+    private static List<String> DLQ_TOPIC_OFFSET_LABEL_NAMES = Arrays.asList(
+            "cluster", "brokerNames", "group", "lastUpdateTimestamp"
+    );
+
+    private void loadTopicOffsetMetric(GaugeMetricFamily family, Map.Entry<ProducerMetric, Double> entry) {
+        family.addMetric(
+                Arrays.asList(
+                        entry.getKey().getClusterName(),
+                        entry.getKey().getBrokerNames(),
+                        entry.getKey().getTopicName(),
+                        String.valueOf(entry.getKey().getLastUpdateTimestamp())
+                ),
+                entry.getValue());
+    }
+
+    private void collectTopicOffsetMetric(List<MetricFamilySamples> mfs) {
+        GaugeMetricFamily topicOffsetF = new GaugeMetricFamily("rocketmq_topic_offset", "TopicOffset", TOPIC_OFFSET_LABEL_NAMES);
+        for (Map.Entry<ProducerMetric, Double> entry : topicOffset.entrySet()) {
+            loadTopicOffsetMetric(topicOffsetF, entry);
+        }
+        mfs.add(topicOffsetF);
+
+        GaugeMetricFamily topicRetryOffsetF = new GaugeMetricFamily("rocketmq_topic_retry_offset", "TopicRetryOffset", TOPIC_OFFSET_LABEL_NAMES);
+        for (Map.Entry<ProducerMetric, Double> entry : topicRetryOffset.entrySet()) {
+            loadTopicOffsetMetric(topicRetryOffsetF, entry);
+        }
+        mfs.add(topicRetryOffsetF);
+
+        GaugeMetricFamily topicDLQOffsetF = new GaugeMetricFamily("rocketmq_topic_dlq_offset", "TopicRetryOffset", DLQ_TOPIC_OFFSET_LABEL_NAMES);
+        for (Map.Entry<DLQTopicOffsetMetric, Double> entry : topicDLQOffset.entrySet()) {
+            topicDLQOffsetF.addMetric(
+                    Arrays.asList(
+                            entry.getKey().getClusterName(),
+                            entry.getKey().getBrokerNames(),
+                            entry.getKey().getGroup(),
+                            String.valueOf(entry.getKey().getLastUpdateTimestamp())
+                    ),
+                    entry.getValue());
+        }
+        mfs.add(topicDLQOffsetF);
+
+    }
 
     @Override
     public List<MetricFamilySamples> collect() {
 
         List<MetricFamilySamples> mfs = new ArrayList<MetricFamilySamples>();
 
-        GaugeMetricFamily topicPutNumsGauge = new GaugeMetricFamily("rocketmq_producer_tps", "TopicPutNums", Arrays.asList("cluster","broker","topic"));
-        for (Map.Entry<ProducerMetric,Double> entry:topicPutNums.entrySet()) {
-            topicPutNumsGauge.addMetric(Arrays.asList(entry.getKey().getClusterName(),entry.getKey().getBrokerName(),entry.getKey().getTopicName()), entry.getValue());
+        collectConsumerMetric(mfs);
+
+        collectTopicOffsetMetric(mfs);
+
+        collectTopicNums(mfs);
+
+        collectGroupNums(mfs);
+
+        collectBrokerNums(mfs);
+
+        collectBrokerRuntimeStats(mfs);
+
+        return mfs;
+    }
+
+    private static List<String> GROUP_PULL_LATENCY_LABEL_NAMES = Arrays.asList(
+            "cluster", "broker", "topic", "group", "queueid"
+    );
+    private static List<String> GROUP_LATENCY_BY_STORETIME_LABEL_NAMES = Arrays.asList(
+            "topic", "group"
+    );
+
+    private static List<String> BROKER_NUMS_LABEL_NAMES = Arrays.asList("cluster", "brokerIP", "brokerHost");
+
+    private static void loadBrokerNums(GaugeMetricFamily family, Map.Entry<BrokerMetric, Double> entry) {
+        family.addMetric(Arrays.asList(
+                entry.getKey().getClusterName(),
+                entry.getKey().getBrokerIP(),
+                entry.getKey().getBrokerHost()),
+                entry.getValue()
+        );
+    }
+
+    private void collectBrokerNums(List<MetricFamilySamples> mfs) {
+        GaugeMetricFamily brokerPutNumsGauge = new GaugeMetricFamily("rocketmq_broker_put_nums", "BrokerPutNums", BROKER_NUMS_LABEL_NAMES);
+        for (Map.Entry<BrokerMetric, Double> entry : brokerPutNums.entrySet()) {
+            loadBrokerNums(brokerPutNumsGauge, entry);
         }
-        mfs.add(topicPutNumsGauge);
+        mfs.add(brokerPutNumsGauge);
 
+        GaugeMetricFamily brokerGetNumsGauge = new GaugeMetricFamily("rocketmq_broker_get_nums", "BrokerGetNums", BROKER_NUMS_LABEL_NAMES);
+        for (Map.Entry<BrokerMetric, Double> entry : brokerGetNums.entrySet()) {
+            loadBrokerNums(brokerGetNumsGauge, entry);
+        }
+        mfs.add(brokerGetNumsGauge);
+    }
+
+
+    private static List<String> GROUP_NUMS_LABEL_NAMES = Arrays.asList(
+            "topic", "group"
+    );
 
-        GaugeMetricFamily topicPutSizeGauge = new GaugeMetricFamily("rocketmq_producer_message_size", "TopicPutMessageSize", Arrays.asList("cluster","broker","topic"));
-        for (Map.Entry<ProducerMetric, Double> entry: topicPutSize.entrySet()) {
-            topicPutSizeGauge.addMetric(Arrays.asList(entry.getKey().getClusterName(),entry.getKey().getBrokerName(),entry.getKey().getTopicName()), entry.getValue());
+    private static <T extends Number> void loadGroupNumsMetric(GaugeMetricFamily family, Map.Entry<ConsumerMetric, T> entry) {
+        family.addMetric(Arrays.asList(
+                entry.getKey().getTopicName(),
+                entry.getKey().getConsumerGroupName()),
+                entry.getValue().doubleValue()
+        );
+    }
+
+    private void collectBrokerRuntimeStatsPutMessageDistributeTime(List<MetricFamilySamples> mfs) {
+        GaugeMetricFamily pmdt0 = new GaugeMetricFamily("rocketmq_brokeruntime_pmdt_0ms", "PutMessageDistributeTimeMap0ms", BROKER_RUNTIME_METRIC_LABEL_NAMES);
+        for (Map.Entry<BrokerRuntimeMetric, Integer> entry : brokerRuntimePutMessageDistributeTimeMap0ms.entrySet()) {
+            loadBrokerRuntimeStatsMetric(pmdt0, entry);
         }
-        mfs.add(topicPutSizeGauge);
+        mfs.add(pmdt0);
 
+        GaugeMetricFamily pmdt0to10ms = new GaugeMetricFamily("rocketmq_brokeruntime_pmdt_0to10ms", "PutMessageDistributeTimeMap0to10ms", BROKER_RUNTIME_METRIC_LABEL_NAMES);
+        for (Map.Entry<BrokerRuntimeMetric, Integer> entry : brokerRuntimePutMessageDistributeTimeMap0to10ms.entrySet()) {
+            loadBrokerRuntimeStatsMetric(pmdt0to10ms, entry);
+        }
+        mfs.add(pmdt0to10ms);
 
-        CounterMetricFamily topicOffsetGauge = new CounterMetricFamily("rocketmq_producer_offset", "TopicOffset", Arrays.asList("cluster","broker","topic"));
-        for (Map.Entry<ProducerMetric, Double> entry: topicOffset.entrySet()) {
-            topicOffsetGauge.addMetric(Arrays.asList(entry.getKey().getClusterName(),entry.getKey().getBrokerName(),entry.getKey().getTopicName()), entry.getValue());
+        GaugeMetricFamily pmdt10to50ms = new GaugeMetricFamily("rocketmq_brokeruntime_pmdt_10to50ms", "PutMessageDistributeTimeMap10to50ms", BROKER_RUNTIME_METRIC_LABEL_NAMES);
+        for (Map.Entry<BrokerRuntimeMetric, Integer> entry : brokerRuntimePutMessageDistributeTimeMap10to50ms.entrySet()) {
+            loadBrokerRuntimeStatsMetric(pmdt10to50ms, entry);
         }
-        mfs.add(topicOffsetGauge);
+        mfs.add(pmdt10to50ms);
 
+        GaugeMetricFamily pmdt50to100ms = new GaugeMetricFamily("rocketmq_brokeruntime_pmdt_50to100ms", "PutMessageDistributeTimeMap50to100ms", BROKER_RUNTIME_METRIC_LABEL_NAMES);
+        for (Map.Entry<BrokerRuntimeMetric, Integer> entry : brokerRuntimePutMessageDistributeTimeMap50to100ms.entrySet()) {
+            loadBrokerRuntimeStatsMetric(pmdt50to100ms, entry);
+        }
+        mfs.add(pmdt50to100ms);
 
-        GaugeMetricFamily brokerPutNumsGauge = new GaugeMetricFamily("rocketmq_broker_tps", "BrokerPutNums", Arrays.asList("cluster","broker"));
-        for (Map.Entry<BrokerMetric, Double> entry: brokerPutNums.entrySet()) {
-            brokerPutNumsGauge.addMetric(Arrays.asList(entry.getKey().getClusterName(),entry.getKey().getBrokerName()), entry.getValue());
+        GaugeMetricFamily pmdt100to200ms = new GaugeMetricFamily("rocketmq_brokeruntime_pmdt_100to200ms", "PutMessageDistributeTimeMap100to200ms", BROKER_RUNTIME_METRIC_LABEL_NAMES);
+        for (Map.Entry<BrokerRuntimeMetric, Integer> entry : brokerRuntimePutMessageDistributeTimeMap100to200ms.entrySet()) {
+            loadBrokerRuntimeStatsMetric(pmdt100to200ms, entry);
         }
-        mfs.add(brokerPutNumsGauge);
+        mfs.add(pmdt100to200ms);
 
+        GaugeMetricFamily pmdt200to500ms = new GaugeMetricFamily("rocketmq_brokeruntime_pmdt_200to500ms", "PutMessageDistributeTimeMap200to500ms", BROKER_RUNTIME_METRIC_LABEL_NAMES);
+        for (Map.Entry<BrokerRuntimeMetric, Integer> entry : brokerRuntimePutMessageDistributeTimeMap200to500ms.entrySet()) {
+            loadBrokerRuntimeStatsMetric(pmdt200to500ms, entry);
+        }
+        mfs.add(pmdt200to500ms);
 
-        GaugeMetricFamily brokerGetNumsGauge = new GaugeMetricFamily("rocketmq_broker_qps", "BrokerGetNums", Arrays.asList("cluster","broker"));
-        for (Map.Entry<BrokerMetric, Double> entry: brokerGetNums.entrySet()) {
-            brokerGetNumsGauge.addMetric(Arrays.asList(entry.getKey().getClusterName(),entry.getKey().getBrokerName()), entry.getValue());
+        GaugeMetricFamily pmdt500to1s = new GaugeMetricFamily("rocketmq_brokeruntime_pmdt_500to1s", "PutMessageDistributeTimeMap500to1s", BROKER_RUNTIME_METRIC_LABEL_NAMES);
+        for (Map.Entry<BrokerRuntimeMetric, Integer> entry : brokerRuntimePutMessageDistributeTimeMap500to1s.entrySet()) {
+            loadBrokerRuntimeStatsMetric(pmdt500to1s, entry);
         }
-        mfs.add(brokerGetNumsGauge);
+        mfs.add(pmdt500to1s);
 
+        GaugeMetricFamily pmdt1to2s = new GaugeMetricFamily("rocketmq_brokeruntime_pmdt_1to2s", "PutMessageDistributeTimeMap1to2s", BROKER_RUNTIME_METRIC_LABEL_NAMES);
+        for (Map.Entry<BrokerRuntimeMetric, Integer> entry : brokerRuntimePutMessageDistributeTimeMap1to2s.entrySet()) {
+            loadBrokerRuntimeStatsMetric(pmdt1to2s, entry);
+        }
+        mfs.add(pmdt1to2s);
 
-        GaugeMetricFamily groupGetNumsGauge = new GaugeMetricFamily("rocketmq_consumer_tps", "GroupGetNums", Arrays.asList("cluster","broker","topic","group"));
-        for (Map.Entry<ConsumerMetric, Double> entry: groupGetNums.entrySet()) {
-            groupGetNumsGauge.addMetric(Arrays.asList(entry.getKey().getClusterName(),entry.getKey().getBrokerName(),entry.getKey().getTopicName(),entry.getKey().getConsumerGroupName()), entry.getValue());
+        GaugeMetricFamily pmdt2to3s = new GaugeMetricFamily("rocketmq_brokeruntime_pmdt_2to3s", "PutMessageDistributeTimeMap2to3s", BROKER_RUNTIME_METRIC_LABEL_NAMES);
+        for (Map.Entry<BrokerRuntimeMetric, Integer> entry : brokerRuntimePutMessageDistributeTimeMap2to3s.entrySet()) {
+            loadBrokerRuntimeStatsMetric(pmdt2to3s, entry);
         }
+        mfs.add(pmdt2to3s);
 
+        GaugeMetricFamily pmdt3to4s = new GaugeMetricFamily("rocketmq_brokeruntime_pmdt_3to4s", "PutMessageDistributeTimeMap3to4s", BROKER_RUNTIME_METRIC_LABEL_NAMES);
+        for (Map.Entry<BrokerRuntimeMetric, Integer> entry : brokerRuntimePutMessageDistributeTimeMap3to4s.entrySet()) {
+            loadBrokerRuntimeStatsMetric(pmdt3to4s, entry);
+        }
+        mfs.add(pmdt3to4s);
+
+        GaugeMetricFamily pmdt4to5s = new GaugeMetricFamily("rocketmq_brokeruntime_pmdt_4to5s", "PutMessageDistributeTimeMap4to5s", BROKER_RUNTIME_METRIC_LABEL_NAMES);
+        for (Map.Entry<BrokerRuntimeMetric, Integer> entry : brokerRuntimePutMessageDistributeTimeMap4to5s.entrySet()) {
+            loadBrokerRuntimeStatsMetric(pmdt4to5s, entry);
+        }
+        mfs.add(pmdt4to5s);
+
+        GaugeMetricFamily pmdt5to10s = new GaugeMetricFamily("rocketmq_brokeruntime_pmdt_5to10s", "PutMessageDistributeTimeMap5to10s", BROKER_RUNTIME_METRIC_LABEL_NAMES);
+        for (Map.Entry<BrokerRuntimeMetric, Integer> entry : brokerRuntimePutMessageDistributeTimeMap5to10s.entrySet()) {
+            loadBrokerRuntimeStatsMetric(pmdt5to10s, entry);
+        }
+        mfs.add(pmdt5to10s);
+
+        GaugeMetricFamily pmdt10stoMore = new GaugeMetricFamily("rocketmq_brokeruntime_pmdt_10stomore", "PutMessageDistributeTimeMap10toMore", BROKER_RUNTIME_METRIC_LABEL_NAMES);
+        for (Map.Entry<BrokerRuntimeMetric, Integer> entry : brokerRuntimePutMessageDistributeTimeMap10toMore.entrySet()) {
+            loadBrokerRuntimeStatsMetric(pmdt10stoMore, entry);
+        }
+        mfs.add(pmdt10stoMore);
+    }
+
+    private void collectGroupNums(List<MetricFamilySamples> mfs) {
+        GaugeMetricFamily groupGetNumsGauge = new GaugeMetricFamily("rocketmq_group_get_nums", "GroupGetNums", GROUP_NUMS_LABEL_NAMES);
+        for (Map.Entry<ConsumerMetric, Double> entry : groupGetNums.entrySet()) {
+            loadGroupNumsMetric(groupGetNumsGauge, entry);
+        }
         mfs.add(groupGetNumsGauge);
 
+        GaugeMetricFamily groupConsumeTPSF = new GaugeMetricFamily("rocketmq_group_consume_tps", "GroupConsumeTPS", GROUP_NUMS_LABEL_NAMES);
+        for (Map.Entry<ConsumerMetric, Double> entry : groupConsumeTPS.entrySet()) {
+            loadGroupNumsMetric(groupConsumeTPSF, entry);
+        }
+        mfs.add(groupConsumeTPSF);
 
-        GaugeMetricFamily groupGetSizeGauge = new GaugeMetricFamily("rocketmq_consumer_message_size", "GroupGetMessageSize", Arrays.asList("cluster","broker","topic","group"));
-        for (Map.Entry<ConsumerMetric, Double> entry: groupGetSize.entrySet()) {
-            groupGetSizeGauge.addMetric(Arrays.asList(entry.getKey().getClusterName(),entry.getKey().getBrokerName(),entry.getKey().getTopicName(),entry.getKey().getConsumerGroupName()), entry.getValue());
+        GaugeMetricFamily groupBrokerTotalOffsetF = new GaugeMetricFamily("rocketmq_group_broker_total_offset", "GroupBrokerTotalOffset", GROUP_NUMS_LABEL_NAMES);
+        for (Map.Entry<ConsumerMetric, Long> entry : groupBrokerTotalOffset.entrySet()) {
+            loadGroupNumsMetric(groupBrokerTotalOffsetF, entry);
         }
-        mfs.add(groupGetSizeGauge);
+        mfs.add(groupBrokerTotalOffsetF);
 
-        CounterMetricFamily groupOffsetGauge = new CounterMetricFamily("rocketmq_consumer_offset", "GroupOffset", Arrays.asList("cluster","broker","topic","group"));
-        for (Map.Entry<ConsumerMetric, Double> entry: groupOffset.entrySet()) {
-            groupOffsetGauge.addMetric(Arrays.asList(entry.getKey().getClusterName(),entry.getKey().getBrokerName(),entry.getKey().getTopicName(),entry.getKey().getConsumerGroupName()), entry.getValue());
+        GaugeMetricFamily groupConsumeTotalOffsetF = new GaugeMetricFamily("rocketmq_group_consume_total_offset", "GroupConsumeTotalOffset", GROUP_NUMS_LABEL_NAMES);
+        for (Map.Entry<ConsumerMetric, Long> entry : groupBrokerTotalOffset.entrySet()) {
+            loadGroupNumsMetric(groupConsumeTotalOffsetF, entry);
         }
-        mfs.add(groupOffsetGauge);
+        mfs.add(groupConsumeTotalOffsetF);
 
+        GaugeMetricFamily groupGetSizeGauge = new GaugeMetricFamily("rocketmq_group_get_messagesize", "GroupGetMessageSize", GROUP_NUMS_LABEL_NAMES);
+        for (Map.Entry<ConsumerMetric, Double> entry : groupGetSize.entrySet()) {
+            loadGroupNumsMetric(groupGetSizeGauge, entry);
+        }
+        mfs.add(groupGetSizeGauge);
 
-        GaugeMetricFamily sendBackNumsGauge = new GaugeMetricFamily("rocketmq_send_back_nums", "SendBackNums", Arrays.asList("cluster","broker","topic","group"));
-        for (Map.Entry<ConsumerMetric, Double> entry: sendBackNums.entrySet()) {
-            sendBackNumsGauge.addMetric(Arrays.asList(entry.getKey().getClusterName(),entry.getKey().getBrokerName(),entry.getKey().getTopicName(),entry.getKey().getConsumerGroupName()), entry.getValue());
+        GaugeMetricFamily sendBackNumsGauge = new GaugeMetricFamily("rocketmq_send_back_nums", "SendBackNums", GROUP_NUMS_LABEL_NAMES);
+        for (Map.Entry<ConsumerMetric, Double> entry : sendBackNums.entrySet()) {
+            loadGroupNumsMetric(sendBackNumsGauge, entry);
         }
         mfs.add(sendBackNumsGauge);
+    }
 
+    private void collectTopicNums(List<MetricFamilySamples> mfs) {
+        GaugeMetricFamily topicPutNumsGauge = new GaugeMetricFamily("rocketmq_topic_put_nums", "TopicPutNums", TOPIC_NUMS_LABEL_NAMES);
+        for (Map.Entry<TopicPutNumMetric, Double> entry : topicPutNums.entrySet()) {
+            loadTopicNumsMetric(topicPutNumsGauge, entry);
+        }
+        mfs.add(topicPutNumsGauge);
 
-        GaugeMetricFamily groupGetLatencyGauge = new GaugeMetricFamily("rocketmq_group_get_latency", "GroupGetLatency", Arrays.asList("cluster","broker","topic","group","queueid"));
-        for (Map.Entry<ConsumerQueueMetric, Double> entry: groupGetLatency.entrySet()) {
-            groupGetLatencyGauge.addMetric(Arrays.asList(entry.getKey().getClusterName(),entry.getKey().getBrokerName(),entry.getKey().getTopicName(),entry.getKey().getConsumerGroupName(),entry.getKey().getQueueId()), entry.getValue());
+        GaugeMetricFamily topicPutSizeGauge = new GaugeMetricFamily("rocketmq_topic_put_messagesize", "TopicPutMessageSize", TOPIC_NUMS_LABEL_NAMES);
+        for (Map.Entry<TopicPutNumMetric, Double> entry : topicPutSize.entrySet()) {
+            loadTopicNumsMetric(topicPutSizeGauge, entry);
         }
-        mfs.add(groupGetLatencyGauge);
+        mfs.add(topicPutSizeGauge);
+    }
 
-        GaugeMetricFamily groupGetLatencyByStoretimeGauge = new GaugeMetricFamily("rocketmq_group_get_latency_by_storetime", "GroupGetLatencyByStoreTime", Arrays.asList("cluster","broker","topic","group"));
-        for (Map.Entry<ConsumerMetric, Double> entry: groupGetLatencyByStoreTime.entrySet()) {
-            groupGetLatencyByStoretimeGauge.addMetric(Arrays.asList(entry.getKey().getClusterName(),entry.getKey().getBrokerName(),entry.getKey().getTopicName(),entry.getKey().getConsumerGroupName()), entry.getValue());
+    private static List<String> TOPIC_NUMS_LABEL_NAMES = Arrays.asList("cluster", "brokers", "topic");
+
+    private void loadTopicNumsMetric(GaugeMetricFamily family, Map.Entry<TopicPutNumMetric, Double> entry) {
+        family.addMetric(
+                Arrays.asList(
+                        entry.getKey().getClusterName(),
+                        entry.getKey().getBrokerNames(),
+                        entry.getKey().getTopicName()
+                ),
+                entry.getValue()
+        );
+    }
+
+    public void addTopicOffsetMetric(String clusterName, String brokerNames, String topic, long lastUpdateTimestamp, double value) {
+        if (topic.startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX)) {
+            topicRetryOffset.put(new ProducerMetric(clusterName, brokerNames, topic, lastUpdateTimestamp), value);
+        } else if (topic.startsWith(MixAll.DLQ_GROUP_TOPIC_PREFIX)) {
+            topicDLQOffset.put(new DLQTopicOffsetMetric(clusterName, brokerNames, topic.replace(MixAll.DLQ_GROUP_TOPIC_PREFIX, ""), lastUpdateTimestamp), value);
+        } else {
+            topicOffset.put(new ProducerMetric(clusterName, brokerNames, topic, lastUpdateTimestamp), value);
         }
-        mfs.add(groupGetLatencyByStoretimeGauge);
+    }
 
-        return mfs;
+    public void addGroupCountMetric(String group, String caddr, String localaddr, int count) {
+        this.consumerCounts.put(new ConsumerCountMetric(group, caddr, localaddr), count);
     }
-    public void AddTopicPutNumsMetric(String clusterName, String brokerName, String topic,  double value)
-    {
-        topicPutNums.put(new ProducerMetric(clusterName,brokerName,topic),value);
+
+    public void addGroupDiffMetric(String countOfOnlineConsumers, String group, String topic, long value) {
+        if (topic.startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX)) {
+            this.consumerRetryDiff.put(new ConsumerTopicDiffMetric(group, topic, countOfOnlineConsumers), value);
+        } else if (topic.startsWith(MixAll.DLQ_GROUP_TOPIC_PREFIX)) {
+            this.consumerDLQDiff.put(new ConsumerTopicDiffMetric(group, topic, countOfOnlineConsumers), value);
+        } else {
+            this.consumerDiff.put(new ConsumerTopicDiffMetric(group, topic, countOfOnlineConsumers), value);
+        }
+    }
+
+    public void addTopicPutNumsMetric(String cluster, String brokerNames, String brokerIP, String brokerHost,
+                                      String topic, double value) {
+        topicPutNums.put(new TopicPutNumMetric(cluster, brokerNames, brokerIP, brokerHost, topic), value);
+    }
+
+    public void addTopicPutSizeMetric(String cluster, String brokerName, String brokerIP, String brokerHost,
+                                      String topic, double value) {
+        topicPutSize.put(new TopicPutNumMetric(cluster, brokerName, brokerIP, brokerHost, topic), value);
+    }
+
+    public void addGroupBrokerTotalOffsetMetric(String topic, String group, long value) {
+        groupBrokerTotalOffset.put(new ConsumerMetric(topic, group), value);
+    }
+
+    public void addGroupConsumerTotalOffsetMetric(String topic, String group, long value) {
+        groupConsumeTotalOffset.put(new ConsumerMetric(topic, group), value);
     }
 
-    public void AddTopicPutSizeMetric(String clusterName, String brokerName, String topic,  double value)
-    {
-        topicPutSize.put(new ProducerMetric(clusterName,brokerName,topic),value);
+    public void addGroupConsumeTPSMetric(String topic, String group, double value) {
+        groupConsumeTPS.put(new ConsumerMetric(topic, group), value);
     }
 
-    public void AddTopicOffsetMetric(String clusterName, String brokerName, String topic,  double value)
-    {
-        topicOffset.put(new ProducerMetric(clusterName,brokerName,topic),value);
+    public void addGroupGetNumsMetric(String topic, String group, double value) {
+        groupGetNums.put(new ConsumerMetric(topic, group), value);
     }
 
-    public void AddBrokerPutNumsMetric(String clusterName, String brokerName,  double value)
-    {
-        brokerPutNums.put(new BrokerMetric(clusterName,brokerName),value);
+    public void addGroupGetSizeMetric(String topic, String group, double value) {
+        groupGetSize.put(new ConsumerMetric(topic, group), value);
     }
 
-    public void AddBrokerGetNumsMetric(String clusterName, String brokerName,  double value)
-    {
-        brokerGetNums.put(new BrokerMetric(clusterName,brokerName),value);
+    public void addSendBackNumsMetric(String topic, String group, double value) {
+        sendBackNums.put(new ConsumerMetric(topic, group), value);
     }
 
-    public void AddGroupGetNumsMetric(String clusterName, String brokerName, String topic, String group,  double value)
-    {
-        groupGetNums.put(new ConsumerMetric(clusterName,brokerName,topic,group),value);
+    public void addBrokerPutNumsMetric(String clusterName, String brokerIP, String brokerHost, double value) {
+        brokerPutNums.put(new BrokerMetric(clusterName, brokerIP, brokerHost), value);
     }
 
-    public void AddGroupGetSizeMetric(String clusterName, String brokerName, String topic, String group,  double value)
-    {
-        groupGetSize.put(new ConsumerMetric(clusterName,brokerName,topic,group),value);
+    public void addBrokerGetNumsMetric(String clusterName, String brokerIP, String brokerHost, double value) {
+        brokerGetNums.put(new BrokerMetric(clusterName, brokerIP, brokerHost), value);
     }
 
-    public void AddGroupOffsetMetric(String clusterName, String brokerName, String topic, String group,  double value)
-    {
-        groupOffset.put(new ConsumerMetric(clusterName,brokerName,topic,group),value);
+    public void addBrokerRuntimeStatsMetric(BrokerRuntimeStats stats, String clusterName, String brokerAddress, String brokerHost) {
+        addBrokerRuntimePutMessageDistributeTimeMap(
+                clusterName, brokerAddress, brokerHost,
+                stats.getBrokerVersionDesc(),
+                stats.getBootTimestamp(),
+                stats.getBrokerVersion(), stats);
+        addCommitLogDirCapacity(clusterName, brokerAddress, brokerHost, stats);
+        addAllKindOfTps(clusterName, brokerAddress, brokerHost, stats);
+
+        brokerRuntimeMsgPutTotalTodayNow.put(new BrokerRuntimeMetric(
+                clusterName, brokerAddress, brokerHost,
+                stats.getBrokerVersionDesc(),
+                stats.getBootTimestamp(),
+                stats.getBrokerVersion()), stats.getMsgPutTotalTodayNow());
+
+        brokerRuntimeMsgGetTotalTodayNow.put(new BrokerRuntimeMetric(
+                clusterName, brokerAddress, brokerHost,
+                stats.getBrokerVersionDesc(),
+                stats.getBootTimestamp(),
+                stats.getBrokerVersion()), stats.getMsgGetTotalTodayNow());
+
+        brokerRuntimeMsgPutTotalTodayMorning.put(new BrokerRuntimeMetric(
+                clusterName, brokerAddress, brokerHost,
+                stats.getBrokerVersionDesc(),
+                stats.getBootTimestamp(),
+                stats.getBrokerVersion()), stats.getMsgPutTotalTodayMorning());
+        brokerRuntimeMsgGetTotalTodayMorning.put(new BrokerRuntimeMetric(
+                clusterName, brokerAddress, brokerHost,
+                stats.getBrokerVersionDesc(),
+                stats.getBootTimestamp(),
+                stats.getBrokerVersion()), stats.getMsgGetTotalTodayMorning());
+        brokerRuntimeMsgPutTotalYesterdayMorning.put(new BrokerRuntimeMetric(
+                clusterName, brokerAddress, brokerHost,
+                stats.getBrokerVersionDesc(),
+                stats.getBootTimestamp(),
+                stats.getBrokerVersion()), stats.getMsgPutTotalYesterdayMorning());
+        brokerRuntimeMsgGetTotalYesterdayMorning.put(new BrokerRuntimeMetric(
+                clusterName, brokerAddress, brokerHost,
+                stats.getBrokerVersionDesc(),
+                stats.getBootTimestamp(),
+                stats.getBrokerVersion()), stats.getMsgGetTotalYesterdayMorning());
+        brokerRuntimeSendThreadPoolQueueHeadWaitTimeMills.put(new BrokerRuntimeMetric(
+                clusterName, brokerAddress, brokerHost,
+                stats.getBrokerVersionDesc(),
+                stats.getBootTimestamp(),
+                stats.getBrokerVersion()), stats.getSendThreadPoolQueueHeadWaitTimeMills());
+        brokerRuntimeQueryThreadPoolQueueHeadWaitTimeMills.put(new BrokerRuntimeMetric(
+                clusterName, brokerAddress, brokerHost,
+                stats.getBrokerVersionDesc(),
+                stats.getBootTimestamp(),
+                stats.getBrokerVersion()), stats.getQueryThreadPoolQueueHeadWaitTimeMills());
+        brokerRuntimePullThreadPoolQueueHeadWaitTimeMills.put(new BrokerRuntimeMetric(
+                clusterName, brokerAddress, brokerHost,
+                stats.getBrokerVersionDesc(),
+                stats.getBootTimestamp(),
+                stats.getBrokerVersion()), stats.getPullThreadPoolQueueHeadWaitTimeMills());
+        brokerRuntimeQueryThreadPoolQueueSize.put(new BrokerRuntimeMetric(
+                clusterName, brokerAddress, brokerHost,
+                stats.getBrokerVersionDesc(),
+                stats.getBootTimestamp(),
+                stats.getBrokerVersion()), stats.getQueryThreadPoolQueueSize());
+        brokerRuntimePullThreadPoolQueueSize.put(new BrokerRuntimeMetric(
+                clusterName, brokerAddress, brokerHost,
+                stats.getBrokerVersionDesc(),
+                stats.getBootTimestamp(),
+                stats.getBrokerVersion()), stats.getPullThreadPoolQueueSize());
+        brokerRuntimeSendThreadPoolQueueCapacity.put(new BrokerRuntimeMetric(
+                clusterName, brokerAddress, brokerHost,
+                stats.getBrokerVersionDesc(),
+                stats.getBootTimestamp(),
+                stats.getBrokerVersion()), stats.getSendThreadPoolQueueCapacity());
+        brokerRuntimePullThreadPoolQueueCapacity.put(new BrokerRuntimeMetric(
+                clusterName, brokerAddress, brokerHost,
+                stats.getBrokerVersionDesc(),
+                stats.getBootTimestamp(),
+                stats.getBrokerVersion()), stats.getPullThreadPoolQueueCapacity());
+
+        brokerRuntimeRemainHowManyDataToFlush.put(new BrokerRuntimeMetric(
+                clusterName, brokerAddress, brokerHost,
+                stats.getBrokerVersionDesc(),
+                stats.getBootTimestamp(),
+                stats.getBrokerVersion()), stats.getRemainHowManyDataToFlush());
+        brokerRuntimeCommitLogMinOffset.put(new BrokerRuntimeMetric(
+                clusterName, brokerAddress, brokerHost,
+                stats.getBrokerVersionDesc(),
+                stats.getBootTimestamp(),
+                stats.getBrokerVersion()), stats.getCommitLogMinOffset());
+        brokerRuntimeCommitLogMaxOffset.put(new BrokerRuntimeMetric(
+                clusterName, brokerAddress, brokerHost,
+                stats.getBrokerVersionDesc(),
+                stats.getBootTimestamp(),
+                stats.getBrokerVersion()), stats.getCommitLogMaxOffset());
+
+
+        brokerRuntimeDispatchMaxBuffer.put(new BrokerRuntimeMetric(
+                clusterName, brokerAddress, brokerHost,
+                stats.getBrokerVersionDesc(),
+                stats.getBootTimestamp(),
+                stats.getBrokerVersion()), stats.getDispatchMaxBuffer());
+        brokerRuntimeConsumeQueueDiskRatio.put(new BrokerRuntimeMetric(
+                clusterName, brokerAddress, brokerHost,
+                stats.getBrokerVersionDesc(),
+                stats.getBootTimestamp(),
+                stats.getBrokerVersion()), stats.getConsumeQueueDiskRatio());
+        brokerRuntimeCommitLogDiskRatio.put(new BrokerRuntimeMetric(
+                clusterName, brokerAddress, brokerHost,
+                stats.getBrokerVersionDesc(),
+                stats.getBootTimestamp(),
+                stats.getBrokerVersion()), stats.getCommitLogDiskRatio());
+        brokerRuntimePageCacheLockTimeMills.put(new BrokerRuntimeMetric(
+                clusterName, brokerAddress, brokerHost,
+                stats.getBrokerVersionDesc(),
+                stats.getBootTimestamp(),
+                stats.getBrokerVersion()), stats.getPageCacheLockTimeMills());
+        brokerRuntimeGetMessageEntireTimeMax.put(new BrokerRuntimeMetric(
+                clusterName, brokerAddress, brokerHost,
+                stats.getBrokerVersionDesc(),
+                stats.getBootTimestamp(),
+                stats.getBrokerVersion()), stats.getGetMessageEntireTimeMax());
+        brokerRuntimePutMessageTimesTotal.put(new BrokerRuntimeMetric(
+                clusterName, brokerAddress, brokerHost,
+                stats.getBrokerVersionDesc(),
+                stats.getBootTimestamp(),
+                stats.getBrokerVersion()), stats.getPutMessageTimesTotal());
+        brokerRuntimeSendThreadPoolQueueSize.put(new BrokerRuntimeMetric(
+                clusterName, brokerAddress, brokerHost,
+                stats.getBrokerVersionDesc(),
+                stats.getBootTimestamp(),
+                stats.getBrokerVersion()), stats.getSendThreadPoolQueueSize());
+        brokerRuntimeStartAcceptSendRequestTimeStamp.put(new BrokerRuntimeMetric(
+                clusterName, brokerAddress, brokerHost,
+                stats.getBrokerVersionDesc(),
+                stats.getBootTimestamp(),
+                stats.getBrokerVersion()), stats.getStartAcceptSendRequestTimeStamp());
+        brokerRuntimePutMessageEntireTimeMax.put(new BrokerRuntimeMetric(
+                clusterName, brokerAddress, brokerHost,
+                stats.getBrokerVersionDesc(),
+                stats.getBootTimestamp(),
+                stats.getBrokerVersion()), stats.getPutMessageEntireTimeMax());
+        brokerRuntimeEarliestMessageTimeStamp.put(new BrokerRuntimeMetric(
+                clusterName, brokerAddress, brokerHost,
+                stats.getBrokerVersionDesc(),
+                stats.getBootTimestamp(),
+                stats.getBrokerVersion()), stats.getEarliestMessageTimeStamp());
+        brokerRuntimeRemainTransientStoreBufferNumbs.put(new BrokerRuntimeMetric(
+                clusterName, brokerAddress, brokerHost,
+                stats.getBrokerVersionDesc(),
+                stats.getBootTimestamp(),
+                stats.getBrokerVersion()), stats.getRemainTransientStoreBufferNumbs());
+        brokerRuntimeQueryThreadPoolQueueCapacity.put(new BrokerRuntimeMetric(
+                clusterName, brokerAddress, brokerHost,
+                stats.getBrokerVersionDesc(),
+                stats.getBootTimestamp(),
+                stats.getBrokerVersion()), stats.getQueryThreadPoolQueueCapacity());
+        brokerRuntimePutMessageAverageSize.put(new BrokerRuntimeMetric(
+                clusterName, brokerAddress, brokerHost,
+                stats.getBrokerVersionDesc(),
+                stats.getBootTimestamp(),
+                stats.getBrokerVersion()), stats.getPutMessageAverageSize());
+        brokerRuntimePutMessageSizeTotal.put(new BrokerRuntimeMetric(
+                clusterName, brokerAddress, brokerHost,
+                stats.getBrokerVersionDesc(),
+                stats.getBootTimestamp(),
+                stats.getBrokerVersion()), stats.getQueryThreadPoolQueueCapacity());
+        brokerRuntimeDispatchBehindBytes.put(new BrokerRuntimeMetric(
+                clusterName, brokerAddress, brokerHost,
+                stats.getBrokerVersionDesc(),
+                stats.getBootTimestamp(),
+                stats.getBrokerVersion()), stats.getDispatchBehindBytes());
     }
 
+    private void addAllKindOfTps(String brokerAddress, String clusterName, String brokerHost, BrokerRuntimeStats stats) {
+        brokerRuntimePutTps10.put(new BrokerRuntimeMetric(
+                clusterName,
+                brokerAddress, brokerHost,
+                stats.getBrokerVersionDesc(),
+                stats.getBootTimestamp(),
+                stats.getBrokerVersion()), stats.getPutTps().getTen());
+        brokerRuntimePutTps60.put(new BrokerRuntimeMetric(
+                clusterName,
+                brokerAddress, brokerHost,
+                stats.getBrokerVersionDesc(),
+                stats.getBootTimestamp(),
+                stats.getBrokerVersion()), stats.getPutTps().getSixty());
+        brokerRuntimePutTps600.put(new BrokerRuntimeMetric(
+                clusterName,
+                brokerAddress, brokerHost,
+                stats.getBrokerVersionDesc(),
+                stats.getBootTimestamp(),
+                stats.getBrokerVersion()), stats.getPutTps().getSixHundred());
+        brokerRuntimeGetMissTps10.put(new BrokerRuntimeMetric(
+                clusterName,
+                brokerAddress, brokerHost,
+                stats.getBrokerVersionDesc(),
+                stats.getBootTimestamp(),
+                stats.getBrokerVersion()), stats.getGetMissTps().getTen());
+        brokerRuntimeGetMissTps60.put(new BrokerRuntimeMetric(
+                clusterName,
+                brokerAddress, brokerHost,
+                stats.getBrokerVersionDesc(),
+                stats.getBootTimestamp(),
+                stats.getBrokerVersion()), stats.getGetMissTps().getSixty());
+        brokerRuntimeGetMissTps600.put(new BrokerRuntimeMetric(
+                clusterName,
+                brokerAddress, brokerHost,
+                stats.getBrokerVersionDesc(),
+                stats.getBootTimestamp(),
+                stats.getBrokerVersion()), stats.getGetMissTps().getSixHundred());
+        brokerRuntimeGetTransferedTps10.put(new BrokerRuntimeMetric(
+                clusterName,
+                brokerAddress, brokerHost,
+                stats.getBrokerVersionDesc(),
+                stats.getBootTimestamp(),
+                stats.getBrokerVersion()), stats.getGetTransferedTps().getTen());
+        brokerRuntimeGetTransferedTps60.put(new BrokerRuntimeMetric(
+                clusterName,
+                brokerAddress, brokerHost,
+                stats.getBrokerVersionDesc(),
+                stats.getBootTimestamp(),
+                stats.getBrokerVersion()), stats.getGetTransferedTps().getSixty());
+        brokerRuntimeGetTransferedTps600.put(new BrokerRuntimeMetric(
+                clusterName,
+                brokerAddress, brokerHost,
+                stats.getBrokerVersionDesc(),
+                stats.getBootTimestamp(),
+                stats.getBrokerVersion()), stats.getGetTransferedTps().getSixHundred());
+        brokerRuntimeGetTotalTps10.put(new BrokerRuntimeMetric(
+                clusterName,
+                brokerAddress, brokerHost,
+                stats.getBrokerVersionDesc(),
+                stats.getBootTimestamp(),
+                stats.getBrokerVersion()), stats.getGetTotalTps().getTen());
+        brokerRuntimeGetTotalTps60.put(new BrokerRuntimeMetric(
+                clusterName,
+                brokerAddress, brokerHost,
+                stats.getBrokerVersionDesc(),
+                stats.getBootTimestamp(),
+                stats.getBrokerVersion()), stats.getGetTotalTps().getSixty());
+        brokerRuntimeGetTotalTps600.put(new BrokerRuntimeMetric(
+                clusterName,
+                brokerAddress, brokerHost,
+                stats.getBrokerVersionDesc(),
+                stats.getBootTimestamp(),
+                stats.getBrokerVersion()), stats.getGetTotalTps().getSixHundred());
+        brokerRuntimeGetFoundTps10.put(new BrokerRuntimeMetric(
+                clusterName,
+                brokerAddress, brokerHost,
+                stats.getBrokerVersionDesc(),
+                stats.getBootTimestamp(),
+                stats.getBrokerVersion()), stats.getGetFoundTps().getTen());
+        brokerRuntimeGetFoundTps60.put(new BrokerRuntimeMetric(
+                clusterName,
+                brokerAddress, brokerHost,
+                stats.getBrokerVersionDesc(),
+                stats.getBootTimestamp(),
+                stats.getBrokerVersion()), stats.getGetFoundTps().getSixty());
+        brokerRuntimeGetFoundTps600.put(new BrokerRuntimeMetric(
+                clusterName,
+                brokerAddress, brokerHost,
+                stats.getBrokerVersionDesc(),
+                stats.getBootTimestamp(),
+                stats.getBrokerVersion()), stats.getGetFoundTps().getSixHundred());
+        brokerRuntimeGetFoundTps600.put(new BrokerRuntimeMetric(
+                clusterName,
+                brokerAddress, brokerHost,
+                stats.getBrokerVersionDesc(),
+                stats.getBootTimestamp(),
+                stats.getBrokerVersion()), stats.getGetFoundTps().getSixHundred());
+    }
 
-    public void AddsendBackNumsMetric(String clusterName, String brokerName, String topic, String group,  double value)
-    {
-        sendBackNums.put(new ConsumerMetric(clusterName,brokerName,topic,group),value);
+    private void addCommitLogDirCapacity(String clusterName, String brokerAddress, String brokerHost, BrokerRuntimeStats stats) {
+        brokerRuntimeCommitLogDirCapacityTotal.put(new BrokerRuntimeMetric(
+                clusterName,
+                brokerAddress, brokerHost,
+                stats.getBrokerVersionDesc(),
+                stats.getBootTimestamp(),
+                stats.getBrokerVersion()), stats.getCommitLogDirCapacityTotal());
+        brokerRuntimeCommitLogDirCapacityFree.put(new BrokerRuntimeMetric(
+                clusterName,
+                brokerAddress, brokerHost,
+                stats.getBrokerVersionDesc(),
+                stats.getBootTimestamp(),
+                stats.getBrokerVersion()), stats.getCommitLogDirCapacityFree());
     }
 
-    public void AddGroupGetLatencyMetric(String clusterName, String brokerName, String topic, String group, String queueId,double value) {
+    private void addBrokerRuntimePutMessageDistributeTimeMap(
+            String clusterName, String brokerAddress, String brokerHost,
+            String brokerDes, long bootTimestamp, int brokerVersion,
+            BrokerRuntimeStats stats) {
+        brokerRuntimePutMessageDistributeTimeMap0ms.put(new BrokerRuntimeMetric(
+                clusterName,
+                brokerAddress, brokerHost,
+                brokerDes,
+                bootTimestamp,
+                brokerVersion), stats.getPutMessageDistributeTimeMap().get("<=0ms"));
+        brokerRuntimePutMessageDistributeTimeMap0to10ms.put(new BrokerRuntimeMetric(
+                clusterName,
+                brokerAddress, brokerHost,
+                brokerDes,
+                bootTimestamp,
+                brokerVersion), stats.getPutMessageDistributeTimeMap().get("0~10ms"));
+        brokerRuntimePutMessageDistributeTimeMap10to50ms.put(new BrokerRuntimeMetric(
+                clusterName,
+                brokerAddress, brokerHost,
+                brokerDes,
+                bootTimestamp,
+                brokerVersion), stats.getPutMessageDistributeTimeMap().get("10~50ms"));
+        brokerRuntimePutMessageDistributeTimeMap50to100ms.put(new BrokerRuntimeMetric(
+                clusterName,
+                brokerAddress, brokerHost,
+                brokerDes,
+                bootTimestamp,
+                brokerVersion), stats.getPutMessageDistributeTimeMap().get("50~100ms"));
+        brokerRuntimePutMessageDistributeTimeMap100to200ms.put(new BrokerRuntimeMetric(
+                clusterName,
+                brokerAddress, brokerHost,
+                brokerDes,
+                bootTimestamp,
+                brokerVersion), stats.getPutMessageDistributeTimeMap().get("100~200ms"));
+        brokerRuntimePutMessageDistributeTimeMap200to500ms.put(new BrokerRuntimeMetric(
+                clusterName,
+                brokerAddress, brokerHost,
+                brokerDes,
+                bootTimestamp,
+                brokerVersion), stats.getPutMessageDistributeTimeMap().get("200~500ms"));
+        brokerRuntimePutMessageDistributeTimeMap500to1s.put(new BrokerRuntimeMetric(
+                clusterName,
+                brokerAddress, brokerHost,
+                brokerDes,
+                bootTimestamp,
+                brokerVersion), stats.getPutMessageDistributeTimeMap().get("500ms~1s"));
+        brokerRuntimePutMessageDistributeTimeMap1to2s.put(new BrokerRuntimeMetric(
+                clusterName,
+                brokerAddress, brokerHost,
+                brokerDes,
+                bootTimestamp,
+                brokerVersion), stats.getPutMessageDistributeTimeMap().get("1~2s"));
+        brokerRuntimePutMessageDistributeTimeMap2to3s.put(new BrokerRuntimeMetric(
+                clusterName,
+                brokerAddress, brokerHost,
+                brokerDes,
+                bootTimestamp,
+                brokerVersion), stats.getPutMessageDistributeTimeMap().get("2~3s"));
+        brokerRuntimePutMessageDistributeTimeMap3to4s.put(new BrokerRuntimeMetric(
+                clusterName,
+                brokerAddress, brokerHost,
+                brokerDes,
+                bootTimestamp,
+                brokerVersion), stats.getPutMessageDistributeTimeMap().get("3~4s"));
+        brokerRuntimePutMessageDistributeTimeMap4to5s.put(new BrokerRuntimeMetric(
+                clusterName,
+                brokerAddress, brokerHost,
+                brokerDes,
+                bootTimestamp,
+                brokerVersion), stats.getPutMessageDistributeTimeMap().get("4~5s"));
+        brokerRuntimePutMessageDistributeTimeMap5to10s.put(new BrokerRuntimeMetric(
+                clusterName,
+                brokerAddress, brokerHost,
+                brokerDes,
+                bootTimestamp,
+                brokerVersion), stats.getPutMessageDistributeTimeMap().get("5~10s"));
+        brokerRuntimePutMessageDistributeTimeMap10toMore.put(new BrokerRuntimeMetric(
+                clusterName,
+                brokerAddress, brokerHost,
+                brokerDes,
+                bootTimestamp,
+                brokerVersion), stats.getPutMessageDistributeTimeMap().get("10s~"));
+    }
 
-        groupGetLatency.put(new ConsumerQueueMetric(clusterName,brokerName,topic,group,queueId),value);
+    private static <T extends Number> void loadBrokerRuntimeStatsMetric(GaugeMetricFamily family, Map.Entry<BrokerRuntimeMetric, T> entry) {
+        family.addMetric(Arrays.asList(
+                entry.getKey().getClusterName(),
+                entry.getKey().getBrokerAddress(),
+                entry.getKey().getBrokerHost(),
+                entry.getKey().getBrokerDes(),
+                String.valueOf(entry.getKey().getBootTimestamp()),
+                String.valueOf(entry.getKey().getBrokerVersion())
+        ), entry.getValue().doubleValue());
     }
 
-    public void AddGroupGetLatencyByStoreTimeMetric(String clusterName, String brokerName, String topic, String group,double value) {
+    private static List<String> BROKER_RUNTIME_METRIC_LABEL_NAMES = Arrays.asList("cluster", "brokerIP", "brokerHost", "des", "boottime", "broker_version");
+
+    private void collectBrokerRuntimeStats(List<MetricFamilySamples> mfs) {
+        collectBrokerRuntimeStatsPutMessageDistributeTime(mfs);
+
+        GaugeMetricFamily brokerRuntimeMsgPutTotalTodayNowF = new GaugeMetricFamily("rocketmq_brokeruntime_msg_put_total_today_now", "brokerRuntimeMsgPutTotalTodayNow", BROKER_RUNTIME_METRIC_LABEL_NAMES);
+        for (Map.Entry<BrokerRuntimeMetric, Long> entry : brokerRuntimeMsgPutTotalTodayNow.entrySet()) {
+            loadBrokerRuntimeStatsMetric(brokerRuntimeMsgPutTotalTodayNowF, entry);
+        }
+        mfs.add(brokerRuntimeMsgPutTotalTodayNowF);
+
+        GaugeMetricFamily brokerRuntimeMsgGetTotalTodayNowF = new GaugeMetricFamily("rocketmq_brokeruntime_msg_gettotal_today_now", "brokerRuntimeMsgGetTotalTodayNow", BROKER_RUNTIME_METRIC_LABEL_NAMES);
+        for (Map.Entry<BrokerRuntimeMetric, Long> entry : brokerRuntimeMsgGetTotalTodayNow.entrySet()) {
+            loadBrokerRuntimeStatsMetric(brokerRuntimeMsgGetTotalTodayNowF, entry);
+        }
+        mfs.add(brokerRuntimeMsgGetTotalTodayNowF);
+
+        GaugeMetricFamily brokerRuntimeDispatchBehindBytesF = new GaugeMetricFamily("rocketmq_brokeruntime_dispatch_behind_bytes", "brokerRuntimeDispatchBehindBytes", BROKER_RUNTIME_METRIC_LABEL_NAMES);
+        for (Map.Entry<BrokerRuntimeMetric, Long> entry : brokerRuntimeDispatchBehindBytes.entrySet()) {
+            loadBrokerRuntimeStatsMetric(brokerRuntimeDispatchBehindBytesF, entry);
+        }
+        mfs.add(brokerRuntimeDispatchBehindBytesF);
+
+        GaugeMetricFamily brokerRuntimePutMessageSizeTotalF = new GaugeMetricFamily("rocketmq_brokeruntime_put_message_size_total", "brokerRuntimePutMessageSizeTotal", BROKER_RUNTIME_METRIC_LABEL_NAMES);
+        for (Map.Entry<BrokerRuntimeMetric, Long> entry : brokerRuntimePutMessageSizeTotal.entrySet()) {
+            loadBrokerRuntimeStatsMetric(brokerRuntimePutMessageSizeTotalF, entry);
+        }
+        mfs.add(brokerRuntimePutMessageSizeTotalF);
+
+        GaugeMetricFamily brokerRuntimePutMessageAverageSizeF = new GaugeMetricFamily("rocketmq_brokeruntime_put_message_average_size", "brokerRuntimePutMessageAverageSize", BROKER_RUNTIME_METRIC_LABEL_NAMES);
+        for (Map.Entry<BrokerRuntimeMetric, Double> entry : brokerRuntimePutMessageAverageSize.entrySet()) {
+            loadBrokerRuntimeStatsMetric(brokerRuntimePutMessageAverageSizeF, entry);
+        }
+        mfs.add(brokerRuntimePutMessageAverageSizeF);
+
+        GaugeMetricFamily brokerRuntimeQueryThreadPoolQueueCapacityF = new GaugeMetricFamily("rocketmq_brokeruntime_query_threadpool_queue_capacity", "brokerRuntimeQueryThreadPoolQueueCapacity", BROKER_RUNTIME_METRIC_LABEL_NAMES);
+        for (Map.Entry<BrokerRuntimeMetric, Long> entry : brokerRuntimeQueryThreadPoolQueueCapacity.entrySet()) {
+            loadBrokerRuntimeStatsMetric(brokerRuntimeQueryThreadPoolQueueCapacityF, entry);
+        }
+        mfs.add(brokerRuntimeQueryThreadPoolQueueCapacityF);
+
+        GaugeMetricFamily brokerRuntimeRemainTransientStoreBufferNumbsF = new GaugeMetricFamily("rocketmq_brokeruntime_remain_transientstore_buffer_numbs", "brokerRuntimeRemainTransientStoreBufferNumbs", BROKER_RUNTIME_METRIC_LABEL_NAMES);
+        for (Map.Entry<BrokerRuntimeMetric, Long> entry : brokerRuntimeRemainTransientStoreBufferNumbs.entrySet()) {
+            loadBrokerRuntimeStatsMetric(brokerRuntimeRemainTransientStoreBufferNumbsF, entry);
+        }
+        mfs.add(brokerRuntimeRemainTransientStoreBufferNumbsF);
+
+        GaugeMetricFamily brokerRuntimeEarliestMessageTimeStampF = new GaugeMetricFamily("rocketmq_brokeruntime_earliest_message_timestamp", "brokerRuntimeEarliestMessageTimeStamp", BROKER_RUNTIME_METRIC_LABEL_NAMES);
+        for (Map.Entry<BrokerRuntimeMetric, Long> entry : brokerRuntimeEarliestMessageTimeStamp.entrySet()) {
+            loadBrokerRuntimeStatsMetric(brokerRuntimeEarliestMessageTimeStampF, entry);
+        }
+        mfs.add(brokerRuntimeEarliestMessageTimeStampF);
+
+        GaugeMetricFamily brokerRuntimePutMessageEntireTimeMaxF = new GaugeMetricFamily("rocketmq_brokeruntime_putmessage_entire_time_max", "brokerRuntimePutMessageEntireTimeMax", BROKER_RUNTIME_METRIC_LABEL_NAMES);
+        for (Map.Entry<BrokerRuntimeMetric, Long> entry : brokerRuntimePutMessageEntireTimeMax.entrySet()) {
+            loadBrokerRuntimeStatsMetric(brokerRuntimePutMessageEntireTimeMaxF, entry);
+        }
+        mfs.add(brokerRuntimePutMessageEntireTimeMaxF);
+
+        GaugeMetricFamily brokerRuntimeStartAcceptSendRequestTimeStampF = new GaugeMetricFamily("rocketmq_brokeruntime_start_accept_sendrequest_time", "brokerRuntimeStartAcceptSendRequestTimeStamp", BROKER_RUNTIME_METRIC_LABEL_NAMES);
+        for (Map.Entry<BrokerRuntimeMetric, Long> entry : brokerRuntimeStartAcceptSendRequestTimeStamp.entrySet()) {
+            loadBrokerRuntimeStatsMetric(brokerRuntimeStartAcceptSendRequestTimeStampF, entry);
+        }
+        mfs.add(brokerRuntimeStartAcceptSendRequestTimeStampF);
 
-        groupGetLatencyByStoreTime.put(new ConsumerMetric(clusterName,brokerName,topic,group),value);
+        GaugeMetricFamily brokerRuntimeSendThreadPoolQueueSizeF = new GaugeMetricFamily("rocketmq_brokeruntime_send_threadpool_queue_size", "brokerRuntimeSendThreadPoolQueueSize", BROKER_RUNTIME_METRIC_LABEL_NAMES);
+        for (Map.Entry<BrokerRuntimeMetric, Long> entry : brokerRuntimeSendThreadPoolQueueSize.entrySet()) {
+            loadBrokerRuntimeStatsMetric(brokerRuntimeSendThreadPoolQueueSizeF, entry);
+        }
+        mfs.add(brokerRuntimeSendThreadPoolQueueSizeF);
+
+        GaugeMetricFamily brokerRuntimePutMessageTimesTotalF = new GaugeMetricFamily("rocketmq_brokeruntime_putmessage_times_total", "brokerRuntimePutMessageTimesTotal", BROKER_RUNTIME_METRIC_LABEL_NAMES);
+        for (Map.Entry<BrokerRuntimeMetric, Long> entry : brokerRuntimePutMessageTimesTotal.entrySet()) {
+            loadBrokerRuntimeStatsMetric(brokerRuntimePutMessageTimesTotalF, entry);
+        }
+        mfs.add(brokerRuntimePutMessageTimesTotalF);
+
+        GaugeMetricFamily brokerRuntimeGetMessageEntireTimeMaxF = new GaugeMetricFamily("rocketmq_brokeruntime_getmessage_entire_time_max", "brokerRuntimeGetMessageEntireTimeMax", BROKER_RUNTIME_METRIC_LABEL_NAMES);
+        for (Map.Entry<BrokerRuntimeMetric, Long> entry : brokerRuntimeGetMessageEntireTimeMax.entrySet()) {
+            loadBrokerRuntimeStatsMetric(brokerRuntimeGetMessageEntireTimeMaxF, entry);
+        }
+        mfs.add(brokerRuntimeGetMessageEntireTimeMaxF);
+
+        GaugeMetricFamily brokerRuntimePageCacheLockTimeMillsF = new GaugeMetricFamily("rocketmq_brokeruntime_pagecache_lock_time_mills", "brokerRuntimePageCacheLockTimeMills", BROKER_RUNTIME_METRIC_LABEL_NAMES);
+        for (Map.Entry<BrokerRuntimeMetric, Long> entry : brokerRuntimePageCacheLockTimeMills.entrySet()) {
+            loadBrokerRuntimeStatsMetric(brokerRuntimePageCacheLockTimeMillsF, entry);
+        }
+        mfs.add(brokerRuntimePageCacheLockTimeMillsF);
+
+        GaugeMetricFamily brokerRuntimeCommitLogDiskRatioF = new GaugeMetricFamily("rocketmq_brokeruntime_commitlog_disk_ratio", "brokerRuntimeCommitLogDiskRatio", BROKER_RUNTIME_METRIC_LABEL_NAMES);
+        for (Map.Entry<BrokerRuntimeMetric, Double> entry : brokerRuntimeCommitLogDiskRatio.entrySet()) {
+            loadBrokerRuntimeStatsMetric(brokerRuntimeCommitLogDiskRatioF, entry);
+        }
+        mfs.add(brokerRuntimeCommitLogDiskRatioF);
+
+        GaugeMetricFamily brokerRuntimeConsumeQueueDiskRatioF = new GaugeMetricFamily("rocketmq_brokeruntime_consumequeue_disk_ratio", "brokerRuntimeConsumeQueueDiskRatio", BROKER_RUNTIME_METRIC_LABEL_NAMES);
+        for (Map.Entry<BrokerRuntimeMetric, Double> entry : brokerRuntimeConsumeQueueDiskRatio.entrySet()) {
+            loadBrokerRuntimeStatsMetric(brokerRuntimeConsumeQueueDiskRatioF, entry);
+        }
+        mfs.add(brokerRuntimeConsumeQueueDiskRatioF);
+
+        GaugeMetricFamily brokerRuntimeGetFoundTps600F = new GaugeMetricFamily("rocketmq_brokeruntime_getfound_tps600", "brokerRuntimeGetFoundTps600", BROKER_RUNTIME_METRIC_LABEL_NAMES);
+        for (Map.Entry<BrokerRuntimeMetric, Double> entry : brokerRuntimeGetFoundTps600.entrySet()) {
+            loadBrokerRuntimeStatsMetric(brokerRuntimeGetFoundTps600F, entry);
+        }
+        mfs.add(brokerRuntimeGetFoundTps600F);
+
+        GaugeMetricFamily brokerRuntimeGetFoundTps60F = new GaugeMetricFamily("rocketmq_brokeruntime_getfound_tps60", "brokerRuntimeGetFoundTps60", BROKER_RUNTIME_METRIC_LABEL_NAMES);
+        for (Map.Entry<BrokerRuntimeMetric, Double> entry : brokerRuntimeGetFoundTps60.entrySet()) {
+            loadBrokerRuntimeStatsMetric(brokerRuntimeGetFoundTps60F, entry);
+        }
+        mfs.add(brokerRuntimeGetFoundTps60F);
+
+        GaugeMetricFamily brokerRuntimeGetFoundTps10F = new GaugeMetricFamily("rocketmq_brokeruntime_getfound_tps10", "brokerRuntimeGetFoundTps10", BROKER_RUNTIME_METRIC_LABEL_NAMES);
+        for (Map.Entry<BrokerRuntimeMetric, Double> entry : brokerRuntimeGetFoundTps10.entrySet()) {
+            loadBrokerRuntimeStatsMetric(brokerRuntimeGetFoundTps10F, entry);
+        }
+        mfs.add(brokerRuntimeGetFoundTps10F);
+
+        GaugeMetricFamily brokerRuntimeGetTotalTps600F = new GaugeMetricFamily("rocketmq_brokeruntime_gettotal_tps600", "brokerRuntimeGetTotalTps600", BROKER_RUNTIME_METRIC_LABEL_NAMES);
+        for (Map.Entry<BrokerRuntimeMetric, Double> entry : brokerRuntimeGetTotalTps600.entrySet()) {
+            loadBrokerRuntimeStatsMetric(brokerRuntimeGetTotalTps600F, entry);
+        }
+        mfs.add(brokerRuntimeGetTotalTps600F);
+
+        GaugeMetricFamily brokerRuntimeGetTotalTps60F = new GaugeMetricFamily("rocketmq_brokeruntime_gettotal_tps60", "brokerRuntimeGetTotalTps60", BROKER_RUNTIME_METRIC_LABEL_NAMES);
+        for (Map.Entry<BrokerRuntimeMetric, Double> entry : brokerRuntimeGetTotalTps60.entrySet()) {
+            loadBrokerRuntimeStatsMetric(brokerRuntimeGetTotalTps60F, entry);
+        }
+        mfs.add(brokerRuntimeGetTotalTps60F);
+
+        GaugeMetricFamily brokerRuntimeGetTotalTps10F = new GaugeMetricFamily("rocketmq_brokeruntime_gettotal_tps10", "brokerRuntimeGetTotalTps10", BROKER_RUNTIME_METRIC_LABEL_NAMES);
+        for (Map.Entry<BrokerRuntimeMetric, Double> entry : brokerRuntimeGetTotalTps10.entrySet()) {
+            loadBrokerRuntimeStatsMetric(brokerRuntimeGetTotalTps10F, entry);
+        }
+        mfs.add(brokerRuntimeGetTotalTps10F);
+
+        GaugeMetricFamily brokerRuntimeGetTransferedTps600F = new GaugeMetricFamily("rocketmq_brokeruntime_gettransfered_tps600", "brokerRuntimeGetTransferedTps600", BROKER_RUNTIME_METRIC_LABEL_NAMES);
+        for (Map.Entry<BrokerRuntimeMetric, Double> entry : brokerRuntimeGetTransferedTps600.entrySet()) {
+            loadBrokerRuntimeStatsMetric(brokerRuntimeGetTransferedTps600F, entry);
+        }
+        mfs.add(brokerRuntimeGetTransferedTps600F);
+
+        GaugeMetricFamily brokerRuntimeGetTransferedTps60F = new GaugeMetricFamily("rocketmq_brokeruntime_gettransfered_tps60", "brokerRuntimeGetTransferedTps60", BROKER_RUNTIME_METRIC_LABEL_NAMES);
+        for (Map.Entry<BrokerRuntimeMetric, Double> entry : brokerRuntimeGetTransferedTps60.entrySet()) {
+            loadBrokerRuntimeStatsMetric(brokerRuntimeGetTransferedTps60F, entry);
+        }
+        mfs.add(brokerRuntimeGetTransferedTps60F);
+
+        GaugeMetricFamily brokerRuntimeGetTransferedTps10F = new GaugeMetricFamily("rocketmq_brokeruntime_gettransfered_tps10", "brokerRuntimeGetTransferedTps10", BROKER_RUNTIME_METRIC_LABEL_NAMES);
+        for (Map.Entry<BrokerRuntimeMetric, Double> entry : brokerRuntimeGetTransferedTps10.entrySet()) {
+            loadBrokerRuntimeStatsMetric(brokerRuntimeGetTransferedTps10F, entry);
+        }
+        mfs.add(brokerRuntimeGetTransferedTps10F);
+
+        GaugeMetricFamily brokerRuntimeGetMissTps600F = new GaugeMetricFamily("rocketmq_brokeruntime_getmiss_tps600", "brokerRuntimeGetMissTps600", BROKER_RUNTIME_METRIC_LABEL_NAMES);
+        for (Map.Entry<BrokerRuntimeMetric, Double> entry : brokerRuntimeGetMissTps600.entrySet()) {
+            loadBrokerRuntimeStatsMetric(brokerRuntimeGetMissTps600F, entry);
+        }
+        mfs.add(brokerRuntimeGetMissTps600F);
+
+        GaugeMetricFamily brokerRuntimeGetMissTps60F = new GaugeMetricFamily("rocketmq_brokeruntime_getmiss_tps60", "brokerRuntimeGetMissTps60", BROKER_RUNTIME_METRIC_LABEL_NAMES);
+        for (Map.Entry<BrokerRuntimeMetric, Double> entry : brokerRuntimeGetMissTps60.entrySet()) {
+            loadBrokerRuntimeStatsMetric(brokerRuntimeGetMissTps60F, entry);
+        }
+        mfs.add(brokerRuntimeGetMissTps60F);
+
+        GaugeMetricFamily brokerRuntimeGetMissTps10F = new GaugeMetricFamily("rocketmq_brokeruntime_getmiss_tps10", "brokerRuntimeGetMissTps10", BROKER_RUNTIME_METRIC_LABEL_NAMES);
+        for (Map.Entry<BrokerRuntimeMetric, Double> entry : brokerRuntimeGetMissTps10.entrySet()) {
+            loadBrokerRuntimeStatsMetric(brokerRuntimeGetMissTps10F, entry);
+        }
+        mfs.add(brokerRuntimeGetMissTps10F);
+
+        GaugeMetricFamily brokerRuntimePutTps600F = new GaugeMetricFamily("rocketmq_brokeruntime_put_tps600", "brokerRuntimePutTps600", BROKER_RUNTIME_METRIC_LABEL_NAMES);
+        for (Map.Entry<BrokerRuntimeMetric, Double> entry : brokerRuntimePutTps600.entrySet()) {
+            loadBrokerRuntimeStatsMetric(brokerRuntimePutTps600F, entry);
+        }
+        mfs.add(brokerRuntimePutTps600F);
+
+        GaugeMetricFamily brokerRuntimePutTps60F = new GaugeMetricFamily("rocketmq_brokeruntime_put_tps60", "brokerRuntimePutTps60", BROKER_RUNTIME_METRIC_LABEL_NAMES);
+        for (Map.Entry<BrokerRuntimeMetric, Double> entry : brokerRuntimePutTps60.entrySet()) {
+            loadBrokerRuntimeStatsMetric(brokerRuntimePutTps60F, entry);
+        }
+        mfs.add(brokerRuntimePutTps60F);
+
+        GaugeMetricFamily brokerRuntimePutTps10F = new GaugeMetricFamily("rocketmq_brokeruntime_put_tps10", "brokerRuntimePutTps10", BROKER_RUNTIME_METRIC_LABEL_NAMES);
+        for (Map.Entry<BrokerRuntimeMetric, Double> entry : brokerRuntimePutTps10.entrySet()) {
+            loadBrokerRuntimeStatsMetric(brokerRuntimePutTps10F, entry);
+        }
+        mfs.add(brokerRuntimePutTps10F);
+
+        GaugeMetricFamily brokerRuntimeDispatchMaxBufferF = new GaugeMetricFamily("rocketmq_brokeruntime_dispatch_maxbuffer", "brokerRuntimeDispatchMaxBuffer", BROKER_RUNTIME_METRIC_LABEL_NAMES);
+        for (Map.Entry<BrokerRuntimeMetric, Long> entry : brokerRuntimeDispatchMaxBuffer.entrySet()) {
+            loadBrokerRuntimeStatsMetric(brokerRuntimeDispatchMaxBufferF, entry);
+        }
+        mfs.add(brokerRuntimeDispatchMaxBufferF);
+
+        GaugeMetricFamily brokerRuntimePullThreadPoolQueueCapacityF = new GaugeMetricFamily("rocketmq_brokeruntime_pull_threadpoolqueue_capacity", "brokerRuntimePullThreadPoolQueueCapacity", BROKER_RUNTIME_METRIC_LABEL_NAMES);
+        for (Map.Entry<BrokerRuntimeMetric, Long> entry : brokerRuntimePullThreadPoolQueueCapacity.entrySet()) {
+            loadBrokerRuntimeStatsMetric(brokerRuntimePullThreadPoolQueueCapacityF, entry);
+        }
+        mfs.add(brokerRuntimePullThreadPoolQueueCapacityF);
+
+        GaugeMetricFamily brokerRuntimeSendThreadPoolQueueCapacityF = new GaugeMetricFamily("rocketmq_brokeruntime_send_threadpoolqueue_capacity", "brokerRuntimeSendThreadPoolQueueCapacity", BROKER_RUNTIME_METRIC_LABEL_NAMES);
+        for (Map.Entry<BrokerRuntimeMetric, Long> entry : brokerRuntimeSendThreadPoolQueueCapacity.entrySet()) {
+            loadBrokerRuntimeStatsMetric(brokerRuntimeSendThreadPoolQueueCapacityF, entry);
+        }
+        mfs.add(brokerRuntimeSendThreadPoolQueueCapacityF);
+
+        GaugeMetricFamily brokerRuntimePullThreadPoolQueueSizeF = new GaugeMetricFamily("rocketmq_brokeruntime_pull_threadpoolqueue_size", "brokerRuntimePullThreadPoolQueueSizeF", BROKER_RUNTIME_METRIC_LABEL_NAMES);
+        for (Map.Entry<BrokerRuntimeMetric, Long> entry : brokerRuntimePullThreadPoolQueueSize.entrySet()) {
+            loadBrokerRuntimeStatsMetric(brokerRuntimePullThreadPoolQueueSizeF, entry);
+        }
+        mfs.add(brokerRuntimePullThreadPoolQueueSizeF);
+
+        GaugeMetricFamily brokerRuntimeQueryThreadPoolQueueSizeF = new GaugeMetricFamily("rocketmq_brokeruntime_query_threadpoolqueue_size", "brokerRuntimeQueryThreadPoolQueueSize", BROKER_RUNTIME_METRIC_LABEL_NAMES);
+        for (Map.Entry<BrokerRuntimeMetric, Long> entry : brokerRuntimeQueryThreadPoolQueueSize.entrySet()) {
+            loadBrokerRuntimeStatsMetric(brokerRuntimeQueryThreadPoolQueueSizeF, entry);
+        }
+        mfs.add(brokerRuntimeQueryThreadPoolQueueSizeF);
+
+        GaugeMetricFamily brokerRuntimePullThreadPoolQueueHeadWaitTimeMillsF = new GaugeMetricFamily("rocketmq_brokeruntime_pull_threadpoolqueue_headwait_timemills", "brokerRuntimePullThreadPoolQueueHeadWaitTimeMills", BROKER_RUNTIME_METRIC_LABEL_NAMES);
+        for (Map.Entry<BrokerRuntimeMetric, Long> entry : brokerRuntimePullThreadPoolQueueHeadWaitTimeMills.entrySet()) {
+            loadBrokerRuntimeStatsMetric(brokerRuntimePullThreadPoolQueueHeadWaitTimeMillsF, entry);
+        }
+        mfs.add(brokerRuntimePullThreadPoolQueueHeadWaitTimeMillsF);
+
+        GaugeMetricFamily brokerRuntimeQueryThreadPoolQueueHeadWaitTimeMillsF = new GaugeMetricFamily("rocketmq_brokeruntime_query_threadpoolqueue_headwait_timemills", "brokerRuntimeQueryThreadPoolQueueHeadWaitTimeMills", BROKER_RUNTIME_METRIC_LABEL_NAMES);
+        for (Map.Entry<BrokerRuntimeMetric, Long> entry : brokerRuntimeQueryThreadPoolQueueHeadWaitTimeMills.entrySet()) {
+            loadBrokerRuntimeStatsMetric(brokerRuntimeQueryThreadPoolQueueHeadWaitTimeMillsF, entry);
+        }
+        mfs.add(brokerRuntimeQueryThreadPoolQueueHeadWaitTimeMillsF);
+
+        GaugeMetricFamily brokerRuntimeSendThreadPoolQueueHeadWaitTimeMillsF = new GaugeMetricFamily("rocketmq_brokeruntime_send_threadpoolqueue_headwait_timemills", "brokerRuntimeSendThreadPoolQueueHeadWaitTimeMills", BROKER_RUNTIME_METRIC_LABEL_NAMES);
+        for (Map.Entry<BrokerRuntimeMetric, Long> entry : brokerRuntimeSendThreadPoolQueueHeadWaitTimeMills.entrySet()) {
+            loadBrokerRuntimeStatsMetric(brokerRuntimeSendThreadPoolQueueHeadWaitTimeMillsF, entry);
+        }
+        mfs.add(brokerRuntimeSendThreadPoolQueueHeadWaitTimeMillsF);
+
+        GaugeMetricFamily brokerRuntimeMsgGetTotalYesterdayMorningF = new GaugeMetricFamily("rocketmq_brokeruntime_msg_gettotal_yesterdaymorning", "brokerRuntimeMsgGetTotalYesterdayMorning", BROKER_RUNTIME_METRIC_LABEL_NAMES);
+        for (Map.Entry<BrokerRuntimeMetric, Long> entry : brokerRuntimeMsgGetTotalYesterdayMorning.entrySet()) {
+            loadBrokerRuntimeStatsMetric(brokerRuntimeMsgGetTotalYesterdayMorningF, entry);
+        }
+        mfs.add(brokerRuntimeMsgGetTotalYesterdayMorningF);
+
+        GaugeMetricFamily brokerRuntimeMsgPutTotalYesterdayMorningF = new GaugeMetricFamily("rocketmq_brokeruntime_msg_puttotal_yesterdaymorning", "brokerRuntimeMsgPutTotalYesterdayMorning", BROKER_RUNTIME_METRIC_LABEL_NAMES);
+        for (Map.Entry<BrokerRuntimeMetric, Long> entry : brokerRuntimeMsgPutTotalYesterdayMorning.entrySet()) {
+            loadBrokerRuntimeStatsMetric(brokerRuntimeMsgPutTotalYesterdayMorningF, entry);
+        }
+        mfs.add(brokerRuntimeMsgPutTotalYesterdayMorningF);
+
+        GaugeMetricFamily brokerRuntimeMsgGetTotalTodayMorningF = new GaugeMetricFamily("rocketmq_brokeruntime_msg_gettotal_todaymorning", "brokerRuntimeMsgGetTotalTodayMorning", BROKER_RUNTIME_METRIC_LABEL_NAMES);
+        for (Map.Entry<BrokerRuntimeMetric, Long> entry : brokerRuntimeMsgGetTotalTodayMorning.entrySet()) {
+            loadBrokerRuntimeStatsMetric(brokerRuntimeMsgGetTotalTodayMorningF, entry);
+        }
+        mfs.add(brokerRuntimeMsgGetTotalTodayMorningF);
+
+        GaugeMetricFamily brokerRuntimeMsgPutTotalTodayMorningF = new GaugeMetricFamily("rocketmq_brokeruntime_msg_puttotal_todaymorning", "brokerRuntimeMsgPutTotalTodayMorning", BROKER_RUNTIME_METRIC_LABEL_NAMES);
+        for (Map.Entry<BrokerRuntimeMetric, Long> entry : brokerRuntimeMsgPutTotalTodayMorning.entrySet()) {
+            loadBrokerRuntimeStatsMetric(brokerRuntimeMsgPutTotalTodayMorningF, entry);
+        }
+        mfs.add(brokerRuntimeMsgPutTotalTodayMorningF);
+
+        GaugeMetricFamily brokerRuntimeCommitLogDirCapacityFreeF = new GaugeMetricFamily("rocketmq_brokeruntime_commitlogdir_capacity_free", "brokerRuntimeCommitLogDirCapacityFree", BROKER_RUNTIME_METRIC_LABEL_NAMES);
+        for (Map.Entry<BrokerRuntimeMetric, Double> entry : brokerRuntimeCommitLogDirCapacityFree.entrySet()) {
+            loadBrokerRuntimeStatsMetric(brokerRuntimeCommitLogDirCapacityFreeF, entry);
+        }
+        mfs.add(brokerRuntimeCommitLogDirCapacityFreeF);
+
+        GaugeMetricFamily brokerRuntimeCommitLogDirCapacityTotalF = new GaugeMetricFamily("rocketmq_brokeruntime_commitlogdir_capacity_total", "brokerRuntimeCommitLogDirCapacityTotal", BROKER_RUNTIME_METRIC_LABEL_NAMES);
+        for (Map.Entry<BrokerRuntimeMetric, Double> entry : brokerRuntimeCommitLogDirCapacityTotal.entrySet()) {
+            loadBrokerRuntimeStatsMetric(brokerRuntimeCommitLogDirCapacityTotalF, entry);
+        }
+        mfs.add(brokerRuntimeCommitLogDirCapacityTotalF);
+
+        GaugeMetricFamily brokerRuntimeCommitLogMaxOffsetF = new GaugeMetricFamily("rocketmq_brokeruntime_commitlog_maxoffset", "brokerRuntimeCommitLogMaxOffset", BROKER_RUNTIME_METRIC_LABEL_NAMES);
+        for (Map.Entry<BrokerRuntimeMetric, Long> entry : brokerRuntimeCommitLogMaxOffset.entrySet()) {
+            loadBrokerRuntimeStatsMetric(brokerRuntimeCommitLogMaxOffsetF, entry);
+        }
+        mfs.add(brokerRuntimeCommitLogMaxOffsetF);
+
+        GaugeMetricFamily brokerRuntimeCommitLogMinOffsetF = new GaugeMetricFamily("rocketmq_brokeruntime_commitlog_minoffset", "brokerRuntimeCommitLogMinOffset", BROKER_RUNTIME_METRIC_LABEL_NAMES);
+        for (Map.Entry<BrokerRuntimeMetric, Long> entry : brokerRuntimeCommitLogMinOffset.entrySet()) {
+            loadBrokerRuntimeStatsMetric(brokerRuntimeCommitLogMinOffsetF, entry);
+        }
+        mfs.add(brokerRuntimeCommitLogMinOffsetF);
+
+        GaugeMetricFamily brokerRuntimeRemainHowManyDataToFlushF = new GaugeMetricFamily("rocketmq_brokeruntime_remain_howmanydata_toflush", "brokerRuntimeRemainHowManyDataToFlush", BROKER_RUNTIME_METRIC_LABEL_NAMES);
+        for (Map.Entry<BrokerRuntimeMetric, Double> entry : brokerRuntimeRemainHowManyDataToFlush.entrySet()) {
+            loadBrokerRuntimeStatsMetric(brokerRuntimeRemainHowManyDataToFlushF, entry);
+        }
+        mfs.add(brokerRuntimeRemainHowManyDataToFlushF);
     }
 }
\ No newline at end of file
diff --git a/src/main/java/org/apache/rocketmq/exporter/config/RMQConfigure.java b/src/main/java/org/apache/rocketmq/exporter/config/RMQConfigure.java
index 2b3956b..e090b68 100644
--- a/src/main/java/org/apache/rocketmq/exporter/config/RMQConfigure.java
+++ b/src/main/java/org/apache/rocketmq/exporter/config/RMQConfigure.java
@@ -23,8 +23,6 @@ import org.slf4j.LoggerFactory;
 import org.springframework.boot.context.properties.ConfigurationProperties;
 import org.springframework.context.annotation.Configuration;
 
-
-
 import static org.apache.rocketmq.client.ClientConfig.SEND_MESSAGE_WITH_VIP_CHANNEL_PROPERTY;
 
 @Configuration
@@ -69,9 +67,11 @@ public class RMQConfigure {
             logger.info("setIsVIPChannel isVIPChannel={}", isVIPChannel);
         }
     }
+
     public boolean isEnableCollect() {
         return enableCollect;
     }
+
     public void setEnableCollect(boolean enableCollect) {
         this.enableCollect = enableCollect;
     }
diff --git a/src/main/java/org/apache/rocketmq/exporter/config/ScheduleConfig.java b/src/main/java/org/apache/rocketmq/exporter/config/ScheduleConfig.java
new file mode 100644
index 0000000..9422272
--- /dev/null
+++ b/src/main/java/org/apache/rocketmq/exporter/config/ScheduleConfig.java
@@ -0,0 +1,26 @@
+package org.apache.rocketmq.exporter.config;
+
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.scheduling.annotation.SchedulingConfigurer;
+import org.springframework.scheduling.config.ScheduledTaskRegistrar;
+
+import java.util.concurrent.Executor;
+import java.util.concurrent.Executors;
+
+@Configuration
+public class ScheduleConfig implements SchedulingConfigurer {
+    @Override
+    public void configureTasks(ScheduledTaskRegistrar taskRegistrar) {
+        taskRegistrar.setScheduler(taskExecutor());
+    }
+
+    @Value("${task.count}")
+    private int taskCount;
+
+    @Bean(destroyMethod = "shutdown")
+    public Executor taskExecutor() {
+        return Executors.newScheduledThreadPool(this.taskCount);
+    }
+}
\ No newline at end of file
diff --git a/src/main/java/org/apache/rocketmq/exporter/controller/RMQMetricsController.java b/src/main/java/org/apache/rocketmq/exporter/controller/RMQMetricsController.java
index 4df69d1..e6fe31c 100644
--- a/src/main/java/org/apache/rocketmq/exporter/controller/RMQMetricsController.java
+++ b/src/main/java/org/apache/rocketmq/exporter/controller/RMQMetricsController.java
@@ -18,8 +18,6 @@ package org.apache.rocketmq.exporter.controller;
 
 
 import org.apache.rocketmq.exporter.service.RMQMetricsService;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
 import org.springframework.web.bind.annotation.RequestMapping;
 import org.springframework.web.bind.annotation.ResponseBody;
@@ -33,19 +31,14 @@ import java.io.StringWriter;
 @RestController
 @EnableAutoConfiguration
 public class RMQMetricsController {
-
-    private final static Logger log = LoggerFactory.getLogger(RMQMetricsController.class);
-
     @Resource
     RMQMetricsService metricsService;
 
     @RequestMapping(value = "${rocketmq.config.webTelemetryPath}")
     @ResponseBody
     private void metrics(HttpServletResponse response) throws IOException {
-
         StringWriter writer = new StringWriter();
-        metricsService.Metrics(writer);
-
+        metricsService.metrics(writer);
         response.setHeader("Content-Type", "text/plain; version=0.0.4; charset=utf-8");
         response.getOutputStream().print(writer.toString());
     }
diff --git a/src/main/java/org/apache/rocketmq/exporter/exception/ServiceException.java b/src/main/java/org/apache/rocketmq/exporter/exception/ServiceException.java
deleted file mode 100644
index ccf3fdc..0000000
--- a/src/main/java/org/apache/rocketmq/exporter/exception/ServiceException.java
+++ /dev/null
@@ -1,31 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- *  Unless required by applicable law or agreed to in writing, software
- *  distributed under the License is distributed on an "AS IS" BASIS,
- *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- *  See the License for the specific language governing permissions and
- *  limitations under the License.
- */
-package org.apache.rocketmq.exporter.exception;
-
-public class ServiceException extends RuntimeException {
-    private static final long serialVersionUID = 9213584003139969215L;
-    private int code;
-
-    public ServiceException(int code, String message) {
-        super(message);
-        this.code = code;
-    }
-
-    public int getCode() {
-        return code;
-    }
-}
diff --git a/src/main/java/org/apache/rocketmq/exporter/model/BrokerRuntimeStats.java b/src/main/java/org/apache/rocketmq/exporter/model/BrokerRuntimeStats.java
new file mode 100644
index 0000000..b6fe3dc
--- /dev/null
+++ b/src/main/java/org/apache/rocketmq/exporter/model/BrokerRuntimeStats.java
@@ -0,0 +1,609 @@
+package org.apache.rocketmq.exporter.model;
+
+import org.apache.rocketmq.common.protocol.body.KVTable;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+public class BrokerRuntimeStats {
+    //今天生产的消息总量
+    private long msgPutTotalTodayNow;
+    //今天消费的消息总量
+    private long msgGetTotalTodayNow;
+
+    //今天早上生产消息总量
+    private long msgPutTotalTodayMorning;
+    //今天早上消费消息总量
+    private long msgGetTotalTodayMorning;
+
+    //昨天早上生产的消息总量
+    private long msgPutTotalYesterdayMorning;
+    //昨天早上消费的消息总量
+    private long msgGetTotalYesterdayMorning;
+
+    //延迟消息位点
+    private List<ScheduleMessageOffsetTable> scheduleMessageOffsetTables = new ArrayList<>();
+
+    //发送线程最大等待时间
+    private long sendThreadPoolQueueHeadWaitTimeMills;
+    //拉取消息线程最大等待时间
+    private long queryThreadPoolQueueHeadWaitTimeMills;
+    //拉取线程最大等待时间
+    private long pullThreadPoolQueueHeadWaitTimeMills;
+
+    //查询线程任务个数
+    private long queryThreadPoolQueueSize;
+    //拉取线程任务个数
+    private long pullThreadPoolQueueSize;
+    //发送线程等待队列长度
+    private long sendThreadPoolQueueCapacity;
+    //拉取线程等待队列长度
+    private long pullThreadPoolQueueCapacity;
+
+    //刷pagecache时间统计
+    private Map<String, Integer> putMessageDistributeTimeMap = new HashMap<>();
+    //还有多少字节的数据没有刷盘
+    private double remainHowManyDataToFlush;
+
+    //commitlog 最小位点
+    private long commitLogMinOffset;
+    //commitlog 最大位点
+    private long commitLogMaxOffset;
+
+    //broker运行时间描述
+    private String runtime;
+    //broker 启动时间
+    private long bootTimestamp;
+    //broker 磁盘总量
+    private double commitLogDirCapacityTotal;
+    //broker 磁盘剩余
+    private double commitLogDirCapacityFree;
+    //broker 版本号
+    private int brokerVersion;
+    //
+    private long dispatchMaxBuffer;
+
+    private PutTps putTps = new PutTps();
+    private GetMissTps getMissTps = new GetMissTps();
+    private GetTransferedTps getTransferedTps = new GetTransferedTps();
+    private GetTotalTps getTotalTps = new GetTotalTps();
+    private GetFoundTps getFoundTps = new GetFoundTps();
+
+    private double consumeQueueDiskRatio;
+    private double commitLogDiskRatio;
+
+    //page cache锁定时间
+    private long pageCacheLockTimeMills;
+
+    private long getMessageEntireTimeMax;
+
+    private long putMessageTimesTotal;
+
+    private String brokerVersionDesc;
+    private long sendThreadPoolQueueSize;
+    private long startAcceptSendRequestTimeStamp;
+    private long putMessageEntireTimeMax;
+    private long earliestMessageTimeStamp;
+
+    private long remainTransientStoreBufferNumbs;
+    private long queryThreadPoolQueueCapacity;
+    //发送消息平均体积大小
+    private double putMessageAverageSize;
+    //全部发送消息数
+    private long putMessageSizeTotal;
+    private long dispatchBehindBytes;
+
+
+    public BrokerRuntimeStats(KVTable kvTable) {
+        this.msgPutTotalTodayNow = Long.parseLong(kvTable.getTable().get("msgPutTotalTodayNow"));
+
+        loadScheduleMessageOffsets(kvTable);
+        loadPutMessageDistributeTime(kvTable.getTable().get("putMessageDistributeTime"));
+
+        loadTps(this.putTps, kvTable.getTable().get("putTps"));
+        loadTps(this.getMissTps, kvTable.getTable().get("getMissTps"));
+        loadTps(this.getTransferedTps, kvTable.getTable().get("getTransferedTps"));
+        loadTps(this.getTotalTps, kvTable.getTable().get("getTotalTps"));
+        loadTps(this.getFoundTps, kvTable.getTable().get("getFoundTps"));
+
+        loadCommitLogDirCapacity(kvTable.getTable().get("commitLogDirCapacity"));
+
+        this.sendThreadPoolQueueHeadWaitTimeMills = Long.parseLong(kvTable.getTable().get("sendThreadPoolQueueHeadWaitTimeMills"));
+        this.queryThreadPoolQueueHeadWaitTimeMills = Long.parseLong(kvTable.getTable().get("queryThreadPoolQueueHeadWaitTimeMills"));
+
+        this.remainHowManyDataToFlush = Double.parseDouble(kvTable.getTable().get("remainHowManyDataToFlush").split(" ")[0]);//byte
+        this.msgGetTotalTodayNow = Long.parseLong(kvTable.getTable().get("msgGetTotalTodayNow"));
+        this.queryThreadPoolQueueSize = Long.parseLong(kvTable.getTable().get("queryThreadPoolQueueSize"));
+        this.bootTimestamp = Long.parseLong(kvTable.getTable().get("bootTimestamp"));
+        this.msgPutTotalYesterdayMorning = Long.parseLong(kvTable.getTable().get("msgPutTotalYesterdayMorning"));
+        this.msgGetTotalYesterdayMorning = Long.parseLong(kvTable.getTable().get("msgGetTotalYesterdayMorning"));
+        this.pullThreadPoolQueueSize = Long.parseLong(kvTable.getTable().get("pullThreadPoolQueueSize"));
+        this.commitLogMinOffset = Long.parseLong(kvTable.getTable().get("commitLogMinOffset"));
+        this.pullThreadPoolQueueHeadWaitTimeMills = Long.parseLong(kvTable.getTable().get("pullThreadPoolQueueHeadWaitTimeMills"));
+        this.runtime = kvTable.getTable().get("runtime");
+        this.dispatchMaxBuffer = Long.parseLong(kvTable.getTable().get("dispatchMaxBuffer"));
+        this.brokerVersion = Integer.parseInt(kvTable.getTable().get("brokerVersion"));
+        this.consumeQueueDiskRatio = Double.parseDouble(kvTable.getTable().get("consumeQueueDiskRatio"));
+        this.pageCacheLockTimeMills = Long.parseLong(kvTable.getTable().get("pageCacheLockTimeMills"));
+        this.commitLogDiskRatio = Double.parseDouble(kvTable.getTable().get("commitLogDiskRatio"));
+        this.commitLogMaxOffset = Long.parseLong(kvTable.getTable().get("commitLogMaxOffset"));
+        this.getMessageEntireTimeMax = Long.parseLong(kvTable.getTable().get("getMessageEntireTimeMax"));
+        this.msgPutTotalTodayMorning = Long.parseLong(kvTable.getTable().get("msgPutTotalTodayMorning"));
+        this.putMessageTimesTotal = Long.parseLong(kvTable.getTable().get("putMessageTimesTotal"));
+        this.msgGetTotalTodayMorning = Long.parseLong(kvTable.getTable().get("msgGetTotalTodayMorning"));
+        this.brokerVersionDesc = kvTable.getTable().get("brokerVersionDesc");
+        this.sendThreadPoolQueueSize = Long.parseLong(kvTable.getTable().get("sendThreadPoolQueueSize"));
+        this.startAcceptSendRequestTimeStamp = Long.parseLong(kvTable.getTable().get("startAcceptSendRequestTimeStamp"));
+        this.putMessageEntireTimeMax = Long.parseLong(kvTable.getTable().get("putMessageEntireTimeMax"));
+        this.earliestMessageTimeStamp = Long.parseLong(kvTable.getTable().get("earliestMessageTimeStamp"));
+        this.remainTransientStoreBufferNumbs = Long.parseLong(kvTable.getTable().get("remainTransientStoreBufferNumbs"));
+        this.queryThreadPoolQueueCapacity = Long.parseLong(kvTable.getTable().get("queryThreadPoolQueueCapacity"));
+        this.putMessageAverageSize = Double.parseDouble(kvTable.getTable().get("putMessageAverageSize"));
+        this.dispatchBehindBytes = Long.parseLong(kvTable.getTable().get("dispatchBehindBytes"));
+        this.putMessageSizeTotal = Long.parseLong(kvTable.getTable().get("putMessageSizeTotal"));
+        this.sendThreadPoolQueueCapacity = Long.parseLong(kvTable.getTable().get("sendThreadPoolQueueCapacity"));
+        this.pullThreadPoolQueueCapacity = Long.parseLong(kvTable.getTable().get("pullThreadPoolQueueCapacity"));
+
+    }
+
+    private void loadCommitLogDirCapacity(String commitLogDirCapacity) {
+        String[] arr = commitLogDirCapacity.split(" ");
+        double total = Double.parseDouble(arr[2]);
+        double free = Double.parseDouble(arr[6]);
+        this.commitLogDirCapacityTotal = total;
+        this.commitLogDirCapacityFree = free;
+    }
+
+    private void loadTps(PutTps putTps, String value) {
+        String[] arr = value.split(" ");
+        putTps.ten = Double.parseDouble(arr[0]);
+        putTps.sixty = Double.parseDouble(arr[1]);
+        putTps.sixHundred = Double.parseDouble(arr[2]);
+    }
+
+    private void loadPutMessageDistributeTime(String str) {
+        String[] arr = str.split(" ");
+        String key = "", value = "";
+        for (String ar : arr) {
+            String[] tarr = ar.split(":");
+            key = tarr[0].replace("[", "").replace("]", "");
+            value = tarr[1];
+            this.putMessageDistributeTimeMap.put(key, Integer.parseInt(value));
+        }
+    }
+
+    public void loadScheduleMessageOffsets(KVTable kvTable) {
+        for (String key : kvTable.getTable().keySet()) {
+            if (key.startsWith("scheduleMessageOffset")) {
+                String[] arr = kvTable.getTable().get(key).split(",");
+                ScheduleMessageOffsetTable table = new ScheduleMessageOffsetTable(
+                        Long.parseLong(arr[0]),
+                        Long.parseLong(arr[1])
+                );
+                this.scheduleMessageOffsetTables.add(table);
+            }
+        }
+    }
+
+    public static class ScheduleMessageOffsetTable {
+        private long delayOffset;
+        private long maxOffset;
+
+        public ScheduleMessageOffsetTable(long first, long second) {
+            this.delayOffset = first;
+            this.maxOffset = second;
+        }
+
+        public long getDelayOffset() {
+            return delayOffset;
+        }
+
+        public void setDelayOffset(long delayOffset) {
+            this.delayOffset = delayOffset;
+        }
+
+        public long getMaxOffset() {
+            return maxOffset;
+        }
+
+        public void setMaxOffset(long maxOffset) {
+            this.maxOffset = maxOffset;
+        }
+    }
+
+    public class PutTps {
+        private double ten;
+        private double sixty;
+        private double sixHundred;
+
+        public double getTen() {
+            return ten;
+        }
+
+        public void setTen(double ten) {
+            this.ten = ten;
+        }
+
+        public double getSixty() {
+            return sixty;
+        }
+
+        public void setSixty(double sixty) {
+            this.sixty = sixty;
+        }
+
+        public double getSixHundred() {
+            return sixHundred;
+        }
+
+        public void setSixHundred(double sixHundred) {
+            this.sixHundred = sixHundred;
+        }
+    }
+
+    public class GetMissTps extends PutTps {
+    }
+
+    public class GetTransferedTps extends PutTps {
+    }
+
+    public class GetTotalTps extends PutTps {
+    }
+
+    public class GetFoundTps extends PutTps {
+    }
+
+    public long getMsgPutTotalTodayNow() {
+        return msgPutTotalTodayNow;
+    }
+
+    public void setMsgPutTotalTodayNow(long msgPutTotalTodayNow) {
+        this.msgPutTotalTodayNow = msgPutTotalTodayNow;
+    }
+
+    public long getMsgGetTotalTodayNow() {
+        return msgGetTotalTodayNow;
+    }
+
+    public void setMsgGetTotalTodayNow(long msgGetTotalTodayNow) {
+        this.msgGetTotalTodayNow = msgGetTotalTodayNow;
+    }
+
+    public long getMsgPutTotalTodayMorning() {
+        return msgPutTotalTodayMorning;
+    }
+
+    public void setMsgPutTotalTodayMorning(long msgPutTotalTodayMorning) {
+        this.msgPutTotalTodayMorning = msgPutTotalTodayMorning;
+    }
+
+    public long getMsgGetTotalTodayMorning() {
+        return msgGetTotalTodayMorning;
+    }
+
+    public void setMsgGetTotalTodayMorning(long msgGetTotalTodayMorning) {
+        this.msgGetTotalTodayMorning = msgGetTotalTodayMorning;
+    }
+
+    public long getMsgPutTotalYesterdayMorning() {
+        return msgPutTotalYesterdayMorning;
+    }
+
+    public void setMsgPutTotalYesterdayMorning(long msgPutTotalYesterdayMorning) {
+        this.msgPutTotalYesterdayMorning = msgPutTotalYesterdayMorning;
+    }
+
+    public long getMsgGetTotalYesterdayMorning() {
+        return msgGetTotalYesterdayMorning;
+    }
+
+    public void setMsgGetTotalYesterdayMorning(long msgGetTotalYesterdayMorning) {
+        this.msgGetTotalYesterdayMorning = msgGetTotalYesterdayMorning;
+    }
+
+    public List<ScheduleMessageOffsetTable> getScheduleMessageOffsetTables() {
+        return scheduleMessageOffsetTables;
+    }
+
+    public void setScheduleMessageOffsetTables(List<ScheduleMessageOffsetTable> scheduleMessageOffsetTables) {
+        this.scheduleMessageOffsetTables = scheduleMessageOffsetTables;
+    }
+
+    public long getSendThreadPoolQueueHeadWaitTimeMills() {
+        return sendThreadPoolQueueHeadWaitTimeMills;
+    }
+
+    public void setSendThreadPoolQueueHeadWaitTimeMills(long sendThreadPoolQueueHeadWaitTimeMills) {
+        this.sendThreadPoolQueueHeadWaitTimeMills = sendThreadPoolQueueHeadWaitTimeMills;
+    }
+
+    public long getQueryThreadPoolQueueHeadWaitTimeMills() {
+        return queryThreadPoolQueueHeadWaitTimeMills;
+    }
+
+    public void setQueryThreadPoolQueueHeadWaitTimeMills(long queryThreadPoolQueueHeadWaitTimeMills) {
+        this.queryThreadPoolQueueHeadWaitTimeMills = queryThreadPoolQueueHeadWaitTimeMills;
+    }
+
+    public long getPullThreadPoolQueueHeadWaitTimeMills() {
+        return pullThreadPoolQueueHeadWaitTimeMills;
+    }
+
+    public void setPullThreadPoolQueueHeadWaitTimeMills(long pullThreadPoolQueueHeadWaitTimeMills) {
+        this.pullThreadPoolQueueHeadWaitTimeMills = pullThreadPoolQueueHeadWaitTimeMills;
+    }
+
+    public long getQueryThreadPoolQueueSize() {
+        return queryThreadPoolQueueSize;
+    }
+
+    public void setQueryThreadPoolQueueSize(long queryThreadPoolQueueSize) {
+        this.queryThreadPoolQueueSize = queryThreadPoolQueueSize;
+    }
+
+    public long getPullThreadPoolQueueSize() {
+        return pullThreadPoolQueueSize;
+    }
+
+    public void setPullThreadPoolQueueSize(long pullThreadPoolQueueSize) {
+        this.pullThreadPoolQueueSize = pullThreadPoolQueueSize;
+    }
+
+    public long getSendThreadPoolQueueCapacity() {
+        return sendThreadPoolQueueCapacity;
+    }
+
+    public void setSendThreadPoolQueueCapacity(long sendThreadPoolQueueCapacity) {
+        this.sendThreadPoolQueueCapacity = sendThreadPoolQueueCapacity;
+    }
+
+    public long getPullThreadPoolQueueCapacity() {
+        return pullThreadPoolQueueCapacity;
+    }
+
+    public void setPullThreadPoolQueueCapacity(long pullThreadPoolQueueCapacity) {
+        this.pullThreadPoolQueueCapacity = pullThreadPoolQueueCapacity;
+    }
+
+    public Map<String, Integer> getPutMessageDistributeTimeMap() {
+        return putMessageDistributeTimeMap;
+    }
+
+    public void setPutMessageDistributeTimeMap(Map<String, Integer> putMessageDistributeTimeMap) {
+        this.putMessageDistributeTimeMap = putMessageDistributeTimeMap;
+    }
+
+    public double getRemainHowManyDataToFlush() {
+        return remainHowManyDataToFlush;
+    }
+
+    public void setRemainHowManyDataToFlush(double remainHowManyDataToFlush) {
+        this.remainHowManyDataToFlush = remainHowManyDataToFlush;
+    }
+
+    public long getCommitLogMinOffset() {
+        return commitLogMinOffset;
+    }
+
+    public void setCommitLogMinOffset(long commitLogMinOffset) {
+        this.commitLogMinOffset = commitLogMinOffset;
+    }
+
+    public long getCommitLogMaxOffset() {
+        return commitLogMaxOffset;
+    }
+
+    public void setCommitLogMaxOffset(long commitLogMaxOffset) {
+        this.commitLogMaxOffset = commitLogMaxOffset;
+    }
+
+    public String getRuntime() {
+        return runtime;
+    }
+
+    public void setRuntime(String runtime) {
+        this.runtime = runtime;
+    }
+
+    public long getBootTimestamp() {
+        return bootTimestamp;
+    }
+
+    public void setBootTimestamp(long bootTimestamp) {
+        this.bootTimestamp = bootTimestamp;
+    }
+
+    public double getCommitLogDirCapacityTotal() {
+        return commitLogDirCapacityTotal;
+    }
+
+    public void setCommitLogDirCapacityTotal(double commitLogDirCapacityTotal) {
+        this.commitLogDirCapacityTotal = commitLogDirCapacityTotal;
+    }
+
+    public double getCommitLogDirCapacityFree() {
+        return commitLogDirCapacityFree;
+    }
+
+    public void setCommitLogDirCapacityFree(double commitLogDirCapacityFree) {
+        this.commitLogDirCapacityFree = commitLogDirCapacityFree;
+    }
+
+    public int getBrokerVersion() {
+        return brokerVersion;
+    }
+
+    public void setBrokerVersion(int brokerVersion) {
+        this.brokerVersion = brokerVersion;
+    }
+
+    public long getDispatchMaxBuffer() {
+        return dispatchMaxBuffer;
+    }
+
+    public void setDispatchMaxBuffer(long dispatchMaxBuffer) {
+        this.dispatchMaxBuffer = dispatchMaxBuffer;
+    }
+
+    public PutTps getPutTps() {
+        return putTps;
+    }
+
+    public void setPutTps(PutTps putTps) {
+        this.putTps = putTps;
+    }
+
+    public GetMissTps getGetMissTps() {
+        return getMissTps;
+    }
+
+    public void setGetMissTps(GetMissTps getMissTps) {
+        this.getMissTps = getMissTps;
+    }
+
+    public GetTransferedTps getGetTransferedTps() {
+        return getTransferedTps;
+    }
+
+    public void setGetTransferedTps(GetTransferedTps getTransferedTps) {
+        this.getTransferedTps = getTransferedTps;
+    }
+
+    public GetTotalTps getGetTotalTps() {
+        return getTotalTps;
+    }
+
+    public void setGetTotalTps(GetTotalTps getTotalTps) {
+        this.getTotalTps = getTotalTps;
+    }
+
+    public GetFoundTps getGetFoundTps() {
+        return getFoundTps;
+    }
+
+    public void setGetFoundTps(GetFoundTps getFoundTps) {
+        this.getFoundTps = getFoundTps;
+    }
+
+    public double getConsumeQueueDiskRatio() {
+        return consumeQueueDiskRatio;
+    }
+
+    public void setConsumeQueueDiskRatio(double consumeQueueDiskRatio) {
+        this.consumeQueueDiskRatio = consumeQueueDiskRatio;
+    }
+
+    public double getCommitLogDiskRatio() {
+        return commitLogDiskRatio;
+    }
+
+    public void setCommitLogDiskRatio(double commitLogDiskRatio) {
+        this.commitLogDiskRatio = commitLogDiskRatio;
+    }
+
+    public long getPageCacheLockTimeMills() {
+        return pageCacheLockTimeMills;
+    }
+
+    public void setPageCacheLockTimeMills(long pageCacheLockTimeMills) {
+        this.pageCacheLockTimeMills = pageCacheLockTimeMills;
+    }
+
+    public long getGetMessageEntireTimeMax() {
+        return getMessageEntireTimeMax;
+    }
+
+    public void setGetMessageEntireTimeMax(long getMessageEntireTimeMax) {
+        this.getMessageEntireTimeMax = getMessageEntireTimeMax;
+    }
+
+    public long getPutMessageTimesTotal() {
+        return putMessageTimesTotal;
+    }
+
+    public void setPutMessageTimesTotal(long putMessageTimesTotal) {
+        this.putMessageTimesTotal = putMessageTimesTotal;
+    }
+
+    public String getBrokerVersionDesc() {
+        return brokerVersionDesc;
+    }
+
+    public void setBrokerVersionDesc(String brokerVersionDesc) {
+        this.brokerVersionDesc = brokerVersionDesc;
+    }
+
+    public long getSendThreadPoolQueueSize() {
+        return sendThreadPoolQueueSize;
+    }
+
+    public void setSendThreadPoolQueueSize(long sendThreadPoolQueueSize) {
+        this.sendThreadPoolQueueSize = sendThreadPoolQueueSize;
+    }
+
+    public long getStartAcceptSendRequestTimeStamp() {
+        return startAcceptSendRequestTimeStamp;
+    }
+
+    public void setStartAcceptSendRequestTimeStamp(long startAcceptSendRequestTimeStamp) {
+        this.startAcceptSendRequestTimeStamp = startAcceptSendRequestTimeStamp;
+    }
+
+    public long getPutMessageEntireTimeMax() {
+        return putMessageEntireTimeMax;
+    }
+
+    public void setPutMessageEntireTimeMax(long putMessageEntireTimeMax) {
+        this.putMessageEntireTimeMax = putMessageEntireTimeMax;
+    }
+
+    public long getEarliestMessageTimeStamp() {
+        return earliestMessageTimeStamp;
+    }
+
+    public void setEarliestMessageTimeStamp(long earliestMessageTimeStamp) {
+        this.earliestMessageTimeStamp = earliestMessageTimeStamp;
+    }
+
+    public long getRemainTransientStoreBufferNumbs() {
+        return remainTransientStoreBufferNumbs;
+    }
+
+    public void setRemainTransientStoreBufferNumbs(long remainTransientStoreBufferNumbs) {
+        this.remainTransientStoreBufferNumbs = remainTransientStoreBufferNumbs;
+    }
+
+    public long getQueryThreadPoolQueueCapacity() {
+        return queryThreadPoolQueueCapacity;
+    }
+
+    public void setQueryThreadPoolQueueCapacity(long queryThreadPoolQueueCapacity) {
+        this.queryThreadPoolQueueCapacity = queryThreadPoolQueueCapacity;
+    }
+
+    public double getPutMessageAverageSize() {
+        return putMessageAverageSize;
+    }
+
+    public void setPutMessageAverageSize(double putMessageAverageSize) {
+        this.putMessageAverageSize = putMessageAverageSize;
+    }
+
+    public long getPutMessageSizeTotal() {
+        return putMessageSizeTotal;
+    }
+
+    public void setPutMessageSizeTotal(long putMessageSizeTotal) {
+        this.putMessageSizeTotal = putMessageSizeTotal;
+    }
+
+    public long getDispatchBehindBytes() {
+        return dispatchBehindBytes;
+    }
+
+    public void setDispatchBehindBytes(long dispatchBehindBytes) {
+        this.dispatchBehindBytes = dispatchBehindBytes;
+    }
+}
diff --git a/src/main/java/org/apache/rocketmq/exporter/model/metrics/BrokerMetric.java b/src/main/java/org/apache/rocketmq/exporter/model/metrics/BrokerMetric.java
index c7a0727..45caff2 100644
--- a/src/main/java/org/apache/rocketmq/exporter/model/metrics/BrokerMetric.java
+++ b/src/main/java/org/apache/rocketmq/exporter/model/metrics/BrokerMetric.java
@@ -17,32 +17,38 @@
 package org.apache.rocketmq.exporter.model.metrics;
 
 public class BrokerMetric {
+    private String clusterName;
+    private String brokerIP;
+    private String brokerHost;
 
-    private  String   clusterName;
-    private  String   brokerName;
-
-
-    public void setClusterName(String cluster) {
-
-        clusterName = cluster;
+    public BrokerMetric(String clusterName, String brokerIP, String brokerHost) {
+        this.clusterName = clusterName;
+        this.brokerIP = brokerIP;
+        this.brokerHost = brokerHost;
     }
-    public  String getClusterName() {
 
+    public String getClusterName() {
         return clusterName;
     }
-    void setBrokerName(String broker) {
 
-        brokerName = broker;
+    public void setClusterName(String clusterName) {
+        this.clusterName = clusterName;
     }
 
-    public String getBrokerName() {
+    public String getBrokerIP() {
+        return brokerIP;
+    }
+
+    public void setBrokerIP(String brokerIP) {
+        this.brokerIP = brokerIP;
+    }
 
-        return brokerName;
+    public String getBrokerHost() {
+        return brokerHost;
     }
 
-    public BrokerMetric(String cluster, String broker) {
-        clusterName = cluster;
-        brokerName  =   broker;
+    public void setBrokerHost(String brokerHost) {
+        this.brokerHost = brokerHost;
     }
 
     @Override
@@ -52,19 +58,20 @@ public class BrokerMetric {
         }
         BrokerMetric other = (BrokerMetric) obj;
 
-        return  other.clusterName.equals(clusterName) && other.brokerName.equals(brokerName);
+        return other.clusterName.equals(clusterName) &&
+                other.brokerIP.equals(brokerIP);
     }
 
     @Override
     public int hashCode() {
         int hash = 1;
         hash = 37 * hash + clusterName.hashCode();
-        hash = 37 * hash + brokerName.hashCode();
+        hash = 37 * hash + brokerIP.hashCode();
         return hash;
     }
 
     @Override
     public String toString() {
-        return "ClusterName: " + clusterName + " BrokerName: " + brokerName;
+        return "ClusterName: " + clusterName + " brokerIP: " + brokerIP + " brokerHost: " + brokerHost;
     }
 }
diff --git a/src/main/java/org/apache/rocketmq/exporter/model/metrics/ConsumerCountMetric.java b/src/main/java/org/apache/rocketmq/exporter/model/metrics/ConsumerCountMetric.java
new file mode 100644
index 0000000..67d9f20
--- /dev/null
+++ b/src/main/java/org/apache/rocketmq/exporter/model/metrics/ConsumerCountMetric.java
@@ -0,0 +1,58 @@
+package org.apache.rocketmq.exporter.model.metrics;
+
+public class ConsumerCountMetric {
+    private String caddr;
+    private String localaddr;
+    private String group;
+
+    public ConsumerCountMetric(String group, String caddr, String localaddr) {
+        this.group = group;
+        this.caddr = caddr;
+        this.localaddr = localaddr;
+    }
+
+    public String getCaddr() {
+        return caddr;
+    }
+
+    public void setCaddr(String caddr) {
+        this.caddr = caddr;
+    }
+
+    public String getLocaladdr() {
+        return localaddr;
+    }
+
+    public void setLocaladdr(String localaddr) {
+        this.localaddr = localaddr;
+    }
+
+    public String getGroup() {
+        return group;
+    }
+
+    public void setGroup(String group) {
+        this.group = group;
+    }
+
+    @Override
+    public boolean equals(Object obj) {
+        if (!(obj instanceof ConsumerCountMetric)) {
+            return false;
+        }
+        ConsumerCountMetric other = (ConsumerCountMetric) obj;
+        return other.group.equals(group);
+    }
+
+    @Override
+    public int hashCode() {
+        int hash = 1;
+        hash = 37 * hash + group.hashCode();
+        return hash;
+    }
+
+    @Override
+    public String toString() {
+        return "group: " + group + " caddr: " + caddr + " localaddr: " + localaddr;
+    }
+}
diff --git a/src/main/java/org/apache/rocketmq/exporter/model/metrics/ConsumerMetric.java b/src/main/java/org/apache/rocketmq/exporter/model/metrics/ConsumerMetric.java
index 9530fff..a0036a2 100644
--- a/src/main/java/org/apache/rocketmq/exporter/model/metrics/ConsumerMetric.java
+++ b/src/main/java/org/apache/rocketmq/exporter/model/metrics/ConsumerMetric.java
@@ -17,32 +17,22 @@
 package org.apache.rocketmq.exporter.model.metrics;
 
 public class ConsumerMetric {
+    private String topicName;
+    private String consumerGroupName;
 
-    private  String   clusterName;
-    private  String   brokerName;
-    private  String   topicName;
-    private  String   consumerGroupName;
-
-    public void setClusterName(String cluster) {
-        clusterName = cluster;
-    }
-    public  String getClusterName() {
-        return clusterName;
-    }
-    void setBrokerName(String broker) {
-        brokerName = broker;
-    }
-
-    public String getBrokerName() {
-        return brokerName;
+    public ConsumerMetric(String topicName, String consumerGroupName) {
+        this.topicName = topicName;
+        this.consumerGroupName = consumerGroupName;
     }
 
-    public void setTopicName(String topic) {
-        topicName = topic;
-    }
     public String getTopicName() {
         return topicName;
     }
+
+    public void setTopicName(String topicName) {
+        this.topicName = topicName;
+    }
+
     public String getConsumerGroupName() {
         return consumerGroupName;
     }
@@ -51,13 +41,6 @@ public class ConsumerMetric {
         this.consumerGroupName = consumerGroupName;
     }
 
-    public ConsumerMetric(String cluster, String broker, String topic,String consumerGroup) {
-        clusterName = cluster;
-        brokerName  =   broker;
-        topicName   =   topic;
-        consumerGroupName   =   consumerGroup;
-    }
-
     @Override
     public boolean equals(Object obj) {
         if (!(obj instanceof ConsumerMetric)) {
@@ -65,15 +48,13 @@ public class ConsumerMetric {
         }
         ConsumerMetric other = (ConsumerMetric) obj;
 
-        return  other.clusterName.equals(clusterName) && other.brokerName.equals(brokerName)
-                && other.topicName.equals(topicName)  && other.consumerGroupName.equals(consumerGroupName);
+        return other.topicName.equals(topicName) &&
+                other.consumerGroupName.equals(consumerGroupName);
     }
 
     @Override
     public int hashCode() {
         int hash = 1;
-        hash = 37 * hash + clusterName.hashCode();
-        hash = 37 * hash + brokerName.hashCode();
         hash = 37 * hash + topicName.hashCode();
         hash = 37 * hash + consumerGroupName.hashCode();
         return hash;
@@ -81,6 +62,6 @@ public class ConsumerMetric {
 
     @Override
     public String toString() {
-        return "ClusterName: " + clusterName + " BrokerName: " + brokerName + " topicName: " + topicName + " ConsumeGroupName: " + consumerGroupName;
+        return "topicName: " + topicName + " ConsumeGroupName: " + consumerGroupName;
     }
 }
diff --git a/src/main/java/org/apache/rocketmq/exporter/model/metrics/ConsumerQueueMetric.java b/src/main/java/org/apache/rocketmq/exporter/model/metrics/ConsumerQueueMetric.java
deleted file mode 100644
index a6453fc..0000000
--- a/src/main/java/org/apache/rocketmq/exporter/model/metrics/ConsumerQueueMetric.java
+++ /dev/null
@@ -1,93 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- *  Unless required by applicable law or agreed to in writing, software
- *  distributed under the License is distributed on an "AS IS" BASIS,
- *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- *  See the License for the specific language governing permissions and
- *  limitations under the License.
- */
-package org.apache.rocketmq.exporter.model.metrics;
-
-public class ConsumerQueueMetric {
-
-    private  String   clusterName;
-    private  String   brokerName;
-    private  String   topicName;
-    private  String   consumerGroupName;
-    private  String   queueId;
-
-    public void setClusterName(String cluster) {
-        clusterName = cluster;
-    }
-    public  String getClusterName() {
-        return clusterName;
-    }
-    void setBrokerName(String broker) {
-        brokerName = broker;
-    }
-    public String getBrokerName() {
-        return brokerName;
-    }
-    public void setTopicName(String topic) {
-        topicName = topic;
-    }
-    public String  getTopicName() {
-        return topicName;
-    }
-    public String getConsumerGroupName() {
-        return consumerGroupName;
-    }
-
-    public void setConsumerGroupName(String consumerGroupName) {
-        this.consumerGroupName = consumerGroupName;
-    }
-    public String getQueueId() {
-        return queueId;
-    }
-    public void setQueueId(String queueId) {
-        this.queueId = queueId;
-    }
-    public ConsumerQueueMetric(String cluster, String broker, String topic, String consumerGroup,String queue) {
-        clusterName = cluster;
-        brokerName  =   broker;
-        topicName   =   topic;
-        consumerGroupName   =   consumerGroup;
-        queueId             =   queue;
-    }
-
-    @Override
-    public boolean equals(Object obj) {
-        if (!(obj instanceof ConsumerQueueMetric)) {
-            return false;
-        }
-        ConsumerQueueMetric other = (ConsumerQueueMetric) obj;
-
-        return  other.clusterName.equals(clusterName) && other.brokerName.equals(brokerName)
-                && other.topicName.equals(topicName)  && other.consumerGroupName.equals(consumerGroupName)
-                && other.queueId.equals(queueId);
-    }
-
-    @Override
-    public int hashCode() {
-        int hash = 1;
-        hash = 37 * hash + clusterName.hashCode();
-        hash = 37 * hash + brokerName.hashCode();
-        hash = 37 * hash + topicName.hashCode();
-        hash = 37 * hash + consumerGroupName.hashCode();
-        hash = 37 * hash + queueId.hashCode();
-        return hash;
-    }
-
-    @Override
-    public String toString() {
-        return "ClusterName: " + clusterName + " BrokerName: " + brokerName + " topicName: " + topicName + " ConsumeGroupName: " + consumerGroupName  +  "queueId: " + queueId;
-    }
-}
diff --git a/src/main/java/org/apache/rocketmq/exporter/model/metrics/ConsumerTopicDiffMetric.java b/src/main/java/org/apache/rocketmq/exporter/model/metrics/ConsumerTopicDiffMetric.java
new file mode 100644
index 0000000..84b15c1
--- /dev/null
+++ b/src/main/java/org/apache/rocketmq/exporter/model/metrics/ConsumerTopicDiffMetric.java
@@ -0,0 +1,61 @@
+package org.apache.rocketmq.exporter.model.metrics;
+
+public class ConsumerTopicDiffMetric {
+    public ConsumerTopicDiffMetric(String group, String topic, String countOfOnlineConsumers) {
+        this.group = group;
+        this.topic = topic;
+        this.countOfOnlineConsumers = countOfOnlineConsumers;
+    }
+
+    private String group;
+    private String topic;
+    private String countOfOnlineConsumers;
+
+    public String getGroup() {
+        return group;
+    }
+
+    public void setGroup(String group) {
+        this.group = group;
+    }
+
+    public String getTopic() {
+        return topic;
+    }
+
+    public void setTopic(String topic) {
+        this.topic = topic;
+    }
+
+    public String getCountOfOnlineConsumers() {
+        return countOfOnlineConsumers;
+    }
+
+    public void setCountOfOnlineConsumers(String countOfOnlineConsumers) {
+        this.countOfOnlineConsumers = countOfOnlineConsumers;
+    }
+
+    @Override
+    public boolean equals(Object obj) {
+        if (!(obj instanceof ConsumerTopicDiffMetric)) {
+            return false;
+        }
+        ConsumerTopicDiffMetric other = (ConsumerTopicDiffMetric) obj;
+
+        return other.group.equals(group) &&
+                other.topic.equals(topic);
+    }
+
+    @Override
+    public int hashCode() {
+        int hash = 1;
+        hash = 37 * hash + group.hashCode();
+        hash = 37 * hash + topic.hashCode();
+        return hash;
+    }
+
+    @Override
+    public String toString() {
+        return "ConsumerGroup: " + group + " Topic: " + topic;
+    }
+}
diff --git a/src/main/java/org/apache/rocketmq/exporter/model/metrics/DLQTopicOffsetMetric.java b/src/main/java/org/apache/rocketmq/exporter/model/metrics/DLQTopicOffsetMetric.java
new file mode 100644
index 0000000..da530de
--- /dev/null
+++ b/src/main/java/org/apache/rocketmq/exporter/model/metrics/DLQTopicOffsetMetric.java
@@ -0,0 +1,72 @@
+package org.apache.rocketmq.exporter.model.metrics;
+
+public class DLQTopicOffsetMetric {
+    private String clusterName;
+    private String brokerNames;
+    private String group;
+    private long lastUpdateTimestamp;
+
+    public DLQTopicOffsetMetric(String clusterName, String brokerNames, String group, long lastUpdateTimestamp) {
+        this.clusterName = clusterName;
+        this.brokerNames = brokerNames;
+        this.group = group;
+        this.lastUpdateTimestamp = lastUpdateTimestamp;
+    }
+
+    public String getClusterName() {
+        return clusterName;
+    }
+
+    public void setClusterName(String clusterName) {
+        this.clusterName = clusterName;
+    }
+
+    public String getBrokerNames() {
+        return brokerNames;
+    }
+
+    public void setBrokerNames(String brokerNames) {
+        this.brokerNames = brokerNames;
+    }
+
+    public String getGroup() {
+        return group;
+    }
+
+    public void setGroup(String group) {
+        this.group = group;
+    }
+
+    public long getLastUpdateTimestamp() {
+        return lastUpdateTimestamp;
+    }
+
+    public void setLastUpdateTimestamp(long lastUpdateTimestamp) {
+        this.lastUpdateTimestamp = lastUpdateTimestamp;
+    }
+
+    @Override
+    public boolean equals(Object obj) {
+        if (!(obj instanceof DLQTopicOffsetMetric)) {
+            return false;
+        }
+        DLQTopicOffsetMetric other = (DLQTopicOffsetMetric) obj;
+
+        return other.clusterName.equals(clusterName) &&
+                other.group.equals(group);
+
+    }
+
+    @Override
+    public int hashCode() {
+        int hash = 1;
+        hash = 37 * hash + clusterName.hashCode();
+        hash = 37 * hash + group.hashCode();
+        return hash;
+    }
+
+    @Override
+    public String toString() {
+        return "ClusterName: " + clusterName + " BrokerNames: " + brokerNames + " group: " + group;
+    }
+}
diff --git a/src/main/java/org/apache/rocketmq/exporter/model/metrics/ProducerMetric.java b/src/main/java/org/apache/rocketmq/exporter/model/metrics/ProducerMetric.java
index 72baa73..fb9c5a7 100644
--- a/src/main/java/org/apache/rocketmq/exporter/model/metrics/ProducerMetric.java
+++ b/src/main/java/org/apache/rocketmq/exporter/model/metrics/ProducerMetric.java
@@ -16,34 +16,50 @@
  */
 package org.apache.rocketmq.exporter.model.metrics;
 
+//每个topic最大位点
 public class ProducerMetric {
+    private String clusterName;
+    private String brokerNames;
+    private String topicName;
+    private long lastUpdateTimestamp;
 
-    private  String   clusterName;
-    private  String   brokerName;
-    private  String   topicName;
-
-    public void setClusterName(String cluster) {
-        clusterName = cluster;
-    }
-    public  String getClusterName() {
+    public String getClusterName() {
         return clusterName;
     }
-    void setBrokerName(String broker) {
-        brokerName = broker;
+
+    public void setClusterName(String clusterName) {
+        this.clusterName = clusterName;
     }
-    public String getBrokerName() {
-        return brokerName;
+
+    public String getBrokerNames() {
+        return brokerNames;
     }
-    public void setTopicName(String topic) {
-        topicName = topic;
+
+    public void setBrokerNames(String brokerNames) {
+        this.brokerNames = brokerNames;
     }
-    public String  getTopicName() {
+
+    public String getTopicName() {
         return topicName;
     }
-    public ProducerMetric(String cluster,String broker,String topic) {
-        clusterName = cluster;
-        brokerName  =   broker;
-        topicName   =   topic;
+
+    public void setTopicName(String topicName) {
+        this.topicName = topicName;
+    }
+
+    public long getLastUpdateTimestamp() {
+        return lastUpdateTimestamp;
+    }
+
+    public void setLastUpdateTimestamp(long lastUpdateTimestamp) {
+        this.lastUpdateTimestamp = lastUpdateTimestamp;
+    }
+
+    public ProducerMetric(String clusterName, String brokerNames, String topicName, long lastUpdateTimestamp) {
+        this.clusterName = clusterName;
+        this.brokerNames = brokerNames;
+        this.topicName = topicName;
+        this.lastUpdateTimestamp = lastUpdateTimestamp;
     }
 
     @Override
@@ -53,21 +69,20 @@ public class ProducerMetric {
         }
         ProducerMetric other = (ProducerMetric) obj;
 
-        return  other.clusterName.equals(clusterName) && other.brokerName.equals(brokerName)
-                && other.topicName.equals(topicName);
+        return other.clusterName.equals(clusterName) &&
+                other.topicName.equals(topicName);
     }
 
     @Override
     public int hashCode() {
         int hash = 1;
         hash = 37 * hash + clusterName.hashCode();
-        hash = 37 * hash + brokerName.hashCode();
         hash = 37 * hash + topicName.hashCode();
         return hash;
     }
 
     @Override
     public String toString() {
-        return "ClusterName: " + clusterName + " BrokerName: " + brokerName + " topicName: " + topicName;
+        return "ClusterName: " + clusterName + " BrokerNames: " + brokerNames + " topicName: " + topicName;
     }
 }
diff --git a/src/main/java/org/apache/rocketmq/exporter/model/metrics/TopicPutNumMetric.java b/src/main/java/org/apache/rocketmq/exporter/model/metrics/TopicPutNumMetric.java
new file mode 100644
index 0000000..37f4e86
--- /dev/null
+++ b/src/main/java/org/apache/rocketmq/exporter/model/metrics/TopicPutNumMetric.java
@@ -0,0 +1,83 @@
+package org.apache.rocketmq.exporter.model.metrics;
+
+public class TopicPutNumMetric {
+    private String clusterName;
+    private String brokerNames;
+    private String brokerIP;
+    private String brokerHost;
+    private String topicName;
+
+    public TopicPutNumMetric(String clusterName, String brokerNames, String brokerIP, String brokerHost, String topicName) {
+        this.clusterName = clusterName;
+        this.brokerNames = brokerNames;
+        this.brokerIP = brokerIP;
+        this.brokerHost = brokerHost;
+        this.topicName = topicName;
+    }
+
+    public String getClusterName() {
+        return clusterName;
+    }
+
+    public void setClusterName(String clusterName) {
+        this.clusterName = clusterName;
+    }
+
+    public String getBrokerNames() {
+        return brokerNames;
+    }
+
+    public void setBrokerNames(String brokerNames) {
+        this.brokerNames = brokerNames;
+    }
+
+    public String getTopicName() {
+        return topicName;
+    }
+
+    public void setTopicName(String topicName) {
+        this.topicName = topicName;
+    }
+
+    public String getBrokerIP() {
+        return brokerIP;
+    }
+
+    public void setBrokerIP(String brokerIP) {
+        this.brokerIP = brokerIP;
+    }
+
+    public String getBrokerHost() {
+        return brokerHost;
+    }
+
+    public void setBrokerHost(String brokerHost) {
+        this.brokerHost = brokerHost;
+    }
+
+    @Override
+    public boolean equals(Object obj) {
+        if (!(obj instanceof TopicPutNumMetric)) {
+            return false;
+        }
+        TopicPutNumMetric other = (TopicPutNumMetric) obj;
+
+        return other.clusterName.equals(clusterName) &&
+                other.brokerIP.equals(brokerIP) &&
+                other.topicName.equals(topicName);
+    }
+
+    @Override
+    public int hashCode() {
+        int hash = 1;
+        hash = 37 * hash + clusterName.hashCode();
+        hash = 37 * hash + topicName.hashCode();
+        hash = 37 * hash + brokerIP.hashCode();
+        return hash;
+    }
+
+    @Override
+    public String toString() {
+        return "ClusterName: " + clusterName + " brokerIP: " + brokerIP + " topicName: " + topicName;
+    }
+}
diff --git a/src/main/java/org/apache/rocketmq/exporter/model/metrics/brokerruntime/BrokerRuntimeMetric.java b/src/main/java/org/apache/rocketmq/exporter/model/metrics/brokerruntime/BrokerRuntimeMetric.java
new file mode 100644
index 0000000..beb583e
--- /dev/null
+++ b/src/main/java/org/apache/rocketmq/exporter/model/metrics/brokerruntime/BrokerRuntimeMetric.java
@@ -0,0 +1,91 @@
+package org.apache.rocketmq.exporter.model.metrics.brokerruntime;
+
+public class BrokerRuntimeMetric {
+    private String clusterName;
+    private String brokerAddress;
+    private String brokerHost;
+    private String brokerDes;
+    private long bootTimestamp;
+    private int brokerVersion;
+
+    public BrokerRuntimeMetric(String clusterName, String brokerAddress, String brokerHost, String brokerDes, long bootTimestamp, int brokerVersion) {
+        this.clusterName = clusterName;
+        this.brokerAddress = brokerAddress;
+        this.brokerHost = brokerHost;
+        this.brokerDes = brokerDes;
+        this.bootTimestamp = bootTimestamp;
+        this.brokerVersion = brokerVersion;
+    }
+
+    public String getClusterName() {
+        return clusterName;
+    }
+
+    public void setClusterName(String clusterName) {
+        this.clusterName = clusterName;
+    }
+
+    public String getBrokerAddress() {
+        return brokerAddress;
+    }
+
+    public void setBrokerAddress(String brokerAddress) {
+        this.brokerAddress = brokerAddress;
+    }
+
+    public String getBrokerHost() {
+        return brokerHost;
+    }
+
+    public void setBrokerHost(String brokerHost) {
+        this.brokerHost = brokerHost;
+    }
+
+    public String getBrokerDes() {
+        return brokerDes;
+    }
+
+    public void setBrokerDes(String brokerDes) {
+        this.brokerDes = brokerDes;
+    }
+
+    public long getBootTimestamp() {
+        return bootTimestamp;
+    }
+
+    public void setBootTimestamp(long bootTimestamp) {
+        this.bootTimestamp = bootTimestamp;
+    }
+
+    public int getBrokerVersion() {
+        return brokerVersion;
+    }
+
+    public void setBrokerVersion(int brokerVersion) {
+        this.brokerVersion = brokerVersion;
+    }
+
+    @Override
+    public boolean equals(Object obj) {
+        if (!(obj instanceof BrokerRuntimeMetric)) {
+            return false;
+        }
+        BrokerRuntimeMetric other = (BrokerRuntimeMetric) obj;
+
+        return other.clusterName.equals(clusterName) &&
+                other.brokerAddress.equals(brokerAddress);
+    }
+
+    @Override
+    public int hashCode() {
+        int hash = 1;
+        hash = 37 * hash + clusterName.hashCode();
+        hash = 37 * hash + brokerAddress.hashCode();
+        return hash;
+    }
+
+    @Override
+    public String toString() {
+        return "ClusterName: " + clusterName + " brokerAddress: " + brokerAddress + " brokerHost: " + brokerHost;
+    }
+}
diff --git a/src/main/java/org/apache/rocketmq/exporter/service/AbstractCommonService.java b/src/main/java/org/apache/rocketmq/exporter/service/AbstractCommonService.java
deleted file mode 100644
index 50f5b0a..0000000
--- a/src/main/java/org/apache/rocketmq/exporter/service/AbstractCommonService.java
+++ /dev/null
@@ -1,50 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- *  Unless required by applicable law or agreed to in writing, software
- *  distributed under the License is distributed on an "AS IS" BASIS,
- *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- *  See the License for the specific language governing permissions and
- *  limitations under the License.
- */
-package org.apache.rocketmq.exporter.service;
-
-import com.google.common.base.Throwables;
-import com.google.common.collect.Sets;
-import org.apache.commons.collections.CollectionUtils;
-import org.apache.rocketmq.tools.admin.MQAdminExt;
-
-import javax.annotation.Resource;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Set;
-
-public abstract class AbstractCommonService {
-    @Resource
-    protected MQAdminExt mqAdminExt;
-    protected final Set<String> changeToBrokerNameSet(HashMap<String, Set<String>> clusterAddrTable,
-        List<String> clusterNameList, List<String> brokerNameList) {
-        Set<String> finalBrokerNameList = Sets.newHashSet();
-        if (CollectionUtils.isNotEmpty(clusterNameList)) {
-            try {
-                for (String clusterName : clusterNameList) {
-                    finalBrokerNameList.addAll(clusterAddrTable.get(clusterName));
-                }
-            }
-            catch (Exception e) {
-                throw Throwables.propagate(e);
-            }
-        }
-        if (CollectionUtils.isNotEmpty(brokerNameList)) {
-            finalBrokerNameList.addAll(brokerNameList);
-        }
-        return finalBrokerNameList;
-    }
-}
diff --git a/src/main/java/org/apache/rocketmq/exporter/service/RMQMetricsService.java b/src/main/java/org/apache/rocketmq/exporter/service/RMQMetricsService.java
index c1f8802..5e2c0d4 100644
--- a/src/main/java/org/apache/rocketmq/exporter/service/RMQMetricsService.java
+++ b/src/main/java/org/apache/rocketmq/exporter/service/RMQMetricsService.java
@@ -18,12 +18,12 @@ package org.apache.rocketmq.exporter.service;
 
 import org.apache.rocketmq.exporter.collector.RMQMetricsCollector;
 
-
 import java.io.IOException;
 import java.io.StringWriter;
 
 
-public interface RMQMetricsService  {
+public interface RMQMetricsService {
     RMQMetricsCollector getCollector();
-    void Metrics(StringWriter writer) throws IOException;
+
+    void metrics(StringWriter writer) throws IOException;
 }
diff --git a/src/main/java/org/apache/rocketmq/exporter/service/client/MQAdminExtImpl.java b/src/main/java/org/apache/rocketmq/exporter/service/client/MQAdminExtImpl.java
index 17ec7d5..f24490c 100644
--- a/src/main/java/org/apache/rocketmq/exporter/service/client/MQAdminExtImpl.java
+++ b/src/main/java/org/apache/rocketmq/exporter/service/client/MQAdminExtImpl.java
@@ -18,10 +18,12 @@ package org.apache.rocketmq.exporter.service.client;
 
 import com.google.common.base.Throwables;
 import org.apache.rocketmq.client.QueryResult;
+import org.apache.rocketmq.client.consumer.DefaultMQPullConsumer;
 import org.apache.rocketmq.client.consumer.PullResult;
 import org.apache.rocketmq.client.exception.MQBrokerException;
 import org.apache.rocketmq.client.exception.MQClientException;
 import org.apache.rocketmq.client.impl.MQAdminImpl;
+import org.apache.rocketmq.client.impl.factory.MQClientInstance;
 import org.apache.rocketmq.common.TopicConfig;
 import org.apache.rocketmq.common.admin.ConsumeStats;
 import org.apache.rocketmq.common.admin.RollbackStats;
@@ -55,11 +57,14 @@ import org.apache.rocketmq.remoting.exception.RemotingException;
 import org.apache.rocketmq.remoting.exception.RemotingSendRequestException;
 import org.apache.rocketmq.remoting.exception.RemotingTimeoutException;
 import org.apache.rocketmq.remoting.protocol.RemotingCommand;
+import org.apache.rocketmq.tools.admin.DefaultMQAdminExt;
 import org.apache.rocketmq.tools.admin.MQAdminExt;
 import org.apache.rocketmq.tools.admin.api.MessageTrack;
 import org.joor.Reflect;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.stereotype.Service;
 
 import java.io.UnsupportedEncodingException;
@@ -70,8 +75,21 @@ import java.util.Set;
 
 import static org.apache.rocketmq.remoting.protocol.RemotingSerializable.decode;
 
-@Service
+@Service("mqAdminExtImpl")
 public class MQAdminExtImpl implements MQAdminExt {
+    @Autowired
+    @Qualifier("defaultMQAdminExt")
+    private DefaultMQAdminExt defaultMQAdminExt;
+
+    @Autowired
+    private DefaultMQPullConsumer pullConsumer;
+
+    @Autowired
+    private RemotingClient remotingClient;
+
+    @Autowired
+    private MQClientInstance mqClientInstance;
+
     private Logger logger = LoggerFactory.getLogger(MQAdminExtImpl.class);
 
     public MQAdminExtImpl() {
@@ -79,37 +97,35 @@ public class MQAdminExtImpl implements MQAdminExt {
 
 
     public PullResult queryMsgByOffset(MessageQueue mq, long offset) throws Exception {
-        return MQAdminInstance.threadLocalMQPullConsumer().pull(mq, "*", offset, 1);
+        return pullConsumer.pull(mq, "*", offset, 1);
     }
 
     @Override
     public void updateBrokerConfig(String brokerAddr, Properties properties)
-        throws RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException,
-        UnsupportedEncodingException, InterruptedException, MQBrokerException {
-        MQAdminInstance.threadLocalMQAdminExt().updateBrokerConfig(brokerAddr, properties);
+            throws RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException,
+            UnsupportedEncodingException, InterruptedException, MQBrokerException {
+        defaultMQAdminExt.updateBrokerConfig(brokerAddr, properties);
     }
 
     @Override
     public void createAndUpdateTopicConfig(String addr, TopicConfig config)
-        throws RemotingException, MQBrokerException, InterruptedException, MQClientException {
-        MQAdminInstance.threadLocalMQAdminExt().createAndUpdateTopicConfig(addr, config);
+            throws RemotingException, MQBrokerException, InterruptedException, MQClientException {
+        defaultMQAdminExt.createAndUpdateTopicConfig(addr, config);
     }
 
     @Override
     public void createAndUpdateSubscriptionGroupConfig(String addr, SubscriptionGroupConfig config)
-        throws RemotingException, MQBrokerException, InterruptedException, MQClientException {
-        MQAdminInstance.threadLocalMQAdminExt().createAndUpdateSubscriptionGroupConfig(addr, config);
+            throws RemotingException, MQBrokerException, InterruptedException, MQClientException {
+        defaultMQAdminExt.createAndUpdateSubscriptionGroupConfig(addr, config);
     }
 
     @Override
     public SubscriptionGroupConfig examineSubscriptionGroupConfig(String addr, String group) {
-        RemotingClient remotingClient = MQAdminInstance.threadLocalRemotingClient();
         RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.GET_ALL_SUBSCRIPTIONGROUP_CONFIG, null);
         RemotingCommand response = null;
         try {
             response = remotingClient.invokeSync(addr, request, 3000);
-        }
-        catch (Exception err) {
+        } catch (Exception err) {
             throw Throwables.propagate(err);
         }
         assert response != null;
@@ -125,13 +141,11 @@ public class MQAdminExtImpl implements MQAdminExt {
 
     @Override
     public TopicConfig examineTopicConfig(String addr, String topic) {
-        RemotingClient remotingClient = MQAdminInstance.threadLocalRemotingClient();
         RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.GET_ALL_TOPIC_CONFIG, null);
         RemotingCommand response = null;
         try {
             response = remotingClient.invokeSync(addr, request, 3000);
-        }
-        catch (Exception err) {
+        } catch (Exception err) {
             throw Throwables.propagate(err);
         }
         switch (response.getCode()) {
@@ -146,238 +160,238 @@ public class MQAdminExtImpl implements MQAdminExt {
 
     @Override
     public TopicStatsTable examineTopicStats(String topic)
-        throws RemotingException, MQClientException, InterruptedException, MQBrokerException {
-        return MQAdminInstance.threadLocalMQAdminExt().examineTopicStats(topic);
+            throws RemotingException, MQClientException, InterruptedException, MQBrokerException {
+        return defaultMQAdminExt.examineTopicStats(topic);
     }
 
     @Override
     public TopicList fetchAllTopicList() throws RemotingException, MQClientException, InterruptedException {
-        TopicList topicList = MQAdminInstance.threadLocalMQAdminExt().fetchAllTopicList();
+        TopicList topicList = defaultMQAdminExt.fetchAllTopicList();
         logger.debug("op=look={}", JsonUtil.obj2String(topicList.getTopicList()));
         return topicList;
     }
 
     @Override
     public KVTable fetchBrokerRuntimeStats(String brokerAddr)
-        throws RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException,
-        InterruptedException, MQBrokerException {
-        return MQAdminInstance.threadLocalMQAdminExt().fetchBrokerRuntimeStats(brokerAddr);
+            throws RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException,
+            InterruptedException, MQBrokerException {
+        return defaultMQAdminExt.fetchBrokerRuntimeStats(brokerAddr);
     }
 
     @Override
     public ConsumeStats examineConsumeStats(String consumerGroup)
-        throws RemotingException, MQClientException, InterruptedException, MQBrokerException {
-        return MQAdminInstance.threadLocalMQAdminExt().examineConsumeStats(consumerGroup);
+            throws RemotingException, MQClientException, InterruptedException, MQBrokerException {
+        return defaultMQAdminExt.examineConsumeStats(consumerGroup);
     }
 
     @Override
     public ConsumeStats examineConsumeStats(String consumerGroup, String topic)
-        throws RemotingException, MQClientException, InterruptedException, MQBrokerException {
-        return MQAdminInstance.threadLocalMQAdminExt().examineConsumeStats(consumerGroup, topic);
+            throws RemotingException, MQClientException, InterruptedException, MQBrokerException {
+        return defaultMQAdminExt.examineConsumeStats(consumerGroup, topic);
     }
 
     @Override
     public ClusterInfo examineBrokerClusterInfo()
-        throws InterruptedException, MQBrokerException, RemotingTimeoutException, RemotingSendRequestException,
-        RemotingConnectException {
-        return MQAdminInstance.threadLocalMQAdminExt().examineBrokerClusterInfo();
+            throws InterruptedException, MQBrokerException, RemotingTimeoutException, RemotingSendRequestException,
+            RemotingConnectException {
+        return defaultMQAdminExt.examineBrokerClusterInfo();
     }
 
     @Override
     public TopicRouteData examineTopicRouteInfo(String topic)
-        throws RemotingException, MQClientException, InterruptedException {
-        return MQAdminInstance.threadLocalMQAdminExt().examineTopicRouteInfo(topic);
+            throws RemotingException, MQClientException, InterruptedException {
+        return defaultMQAdminExt.examineTopicRouteInfo(topic);
     }
 
     @Override
     public ConsumerConnection examineConsumerConnectionInfo(String consumerGroup)
-        throws RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException,
-        InterruptedException, MQBrokerException, RemotingException, MQClientException {
-        return MQAdminInstance.threadLocalMQAdminExt().examineConsumerConnectionInfo(consumerGroup);
+            throws
+            InterruptedException, MQBrokerException, RemotingException, MQClientException {
+        return defaultMQAdminExt.examineConsumerConnectionInfo(consumerGroup);
     }
 
     @Override
     public ProducerConnection examineProducerConnectionInfo(String producerGroup, String topic)
-        throws RemotingException, MQClientException, InterruptedException, MQBrokerException {
-        return MQAdminInstance.threadLocalMQAdminExt().examineProducerConnectionInfo(producerGroup, topic);
+            throws RemotingException, MQClientException, InterruptedException, MQBrokerException {
+        return defaultMQAdminExt.examineProducerConnectionInfo(producerGroup, topic);
     }
 
     @Override
     public List<String> getNameServerAddressList() {
-        return MQAdminInstance.threadLocalMQAdminExt().getNameServerAddressList();
+        return defaultMQAdminExt.getNameServerAddressList();
     }
 
     @Override
     public int wipeWritePermOfBroker(String namesrvAddr, String brokerName)
-        throws RemotingCommandException, RemotingConnectException, RemotingSendRequestException,
-        RemotingTimeoutException, InterruptedException, MQClientException {
-        return MQAdminInstance.threadLocalMQAdminExt().wipeWritePermOfBroker(namesrvAddr, brokerName);
+            throws RemotingCommandException, RemotingConnectException, RemotingSendRequestException,
+            RemotingTimeoutException, InterruptedException, MQClientException {
+        return defaultMQAdminExt.wipeWritePermOfBroker(namesrvAddr, brokerName);
     }
 
     @Override
     public void putKVConfig(String namespace, String key, String value) {
-        MQAdminInstance.threadLocalMQAdminExt().putKVConfig(namespace, key, value);
+        defaultMQAdminExt.putKVConfig(namespace, key, value);
     }
 
     @Override
     public String getKVConfig(String namespace, String key)
-        throws RemotingException, MQClientException, InterruptedException {
-        return MQAdminInstance.threadLocalMQAdminExt().getKVConfig(namespace, key);
+            throws RemotingException, MQClientException, InterruptedException {
+        return defaultMQAdminExt.getKVConfig(namespace, key);
     }
 
     @Override
     public KVTable getKVListByNamespace(String namespace)
-        throws RemotingException, MQClientException, InterruptedException {
-        return MQAdminInstance.threadLocalMQAdminExt().getKVListByNamespace(namespace);
+            throws RemotingException, MQClientException, InterruptedException {
+        return defaultMQAdminExt.getKVListByNamespace(namespace);
     }
 
     @Override
     public void deleteTopicInBroker(Set<String> addrs, String topic)
-        throws RemotingException, MQBrokerException, InterruptedException, MQClientException {
+            throws RemotingException, MQBrokerException, InterruptedException, MQClientException {
         logger.info("addrs={} topic={}", JsonUtil.obj2String(addrs), topic);
-        MQAdminInstance.threadLocalMQAdminExt().deleteTopicInBroker(addrs, topic);
+        defaultMQAdminExt.deleteTopicInBroker(addrs, topic);
     }
 
     @Override
     public void deleteTopicInNameServer(Set<String> addrs, String topic)
-        throws RemotingException, MQBrokerException, InterruptedException, MQClientException {
-        MQAdminInstance.threadLocalMQAdminExt().deleteTopicInNameServer(addrs, topic);
+            throws RemotingException, MQBrokerException, InterruptedException, MQClientException {
+        defaultMQAdminExt.deleteTopicInNameServer(addrs, topic);
     }
 
     @Override
     public void deleteSubscriptionGroup(String addr, String groupName)
-        throws RemotingException, MQBrokerException, InterruptedException, MQClientException {
-        MQAdminInstance.threadLocalMQAdminExt().deleteSubscriptionGroup(addr, groupName);
+            throws RemotingException, MQBrokerException, InterruptedException, MQClientException {
+        defaultMQAdminExt.deleteSubscriptionGroup(addr, groupName);
     }
 
     @Override
     public void createAndUpdateKvConfig(String namespace, String key, String value)
-        throws RemotingException, MQBrokerException, InterruptedException, MQClientException {
-        MQAdminInstance.threadLocalMQAdminExt().createAndUpdateKvConfig(namespace, key, value);
+            throws RemotingException, MQBrokerException, InterruptedException, MQClientException {
+        defaultMQAdminExt.createAndUpdateKvConfig(namespace, key, value);
     }
 
     @Override
     public void deleteKvConfig(String namespace, String key)
-        throws RemotingException, MQBrokerException, InterruptedException, MQClientException {
-        MQAdminInstance.threadLocalMQAdminExt().deleteKvConfig(namespace, key);
+            throws RemotingException, MQBrokerException, InterruptedException, MQClientException {
+        defaultMQAdminExt.deleteKvConfig(namespace, key);
     }
 
     @Override
     public List<RollbackStats> resetOffsetByTimestampOld(String consumerGroup, String topic, long timestamp,
-        boolean force) throws RemotingException, MQBrokerException, InterruptedException, MQClientException {
-        return MQAdminInstance.threadLocalMQAdminExt().resetOffsetByTimestampOld(consumerGroup, topic, timestamp, force);
+                                                         boolean force) throws RemotingException, MQBrokerException, InterruptedException, MQClientException {
+        return defaultMQAdminExt.resetOffsetByTimestampOld(consumerGroup, topic, timestamp, force);
     }
 
     @Override
     public Map<MessageQueue, Long> resetOffsetByTimestamp(String topic, String group, long timestamp,
-        boolean isForce) throws RemotingException, MQBrokerException, InterruptedException, MQClientException {
-        return MQAdminInstance.threadLocalMQAdminExt().resetOffsetByTimestamp(topic, group, timestamp, isForce);
+                                                          boolean isForce) throws RemotingException, MQBrokerException, InterruptedException, MQClientException {
+        return defaultMQAdminExt.resetOffsetByTimestamp(topic, group, timestamp, isForce);
     }
 
     @Override
     public void resetOffsetNew(String consumerGroup, String topic, long timestamp)
-        throws RemotingException, MQBrokerException, InterruptedException, MQClientException {
-        MQAdminInstance.threadLocalMQAdminExt().resetOffsetNew(consumerGroup, topic, timestamp);
+            throws RemotingException, MQBrokerException, InterruptedException, MQClientException {
+        defaultMQAdminExt.resetOffsetNew(consumerGroup, topic, timestamp);
     }
 
     @Override
     public Map<String, Map<MessageQueue, Long>> getConsumeStatus(String topic, String group,
-        String clientAddr) throws RemotingException, MQBrokerException, InterruptedException, MQClientException {
-        return MQAdminInstance.threadLocalMQAdminExt().getConsumeStatus(topic, group, clientAddr);
+                                                                 String clientAddr) throws RemotingException, MQBrokerException, InterruptedException, MQClientException {
+        return defaultMQAdminExt.getConsumeStatus(topic, group, clientAddr);
     }
 
     @Override
     public void createOrUpdateOrderConf(String key, String value, boolean isCluster)
-        throws RemotingException, MQBrokerException, InterruptedException, MQClientException {
-        MQAdminInstance.threadLocalMQAdminExt().createOrUpdateOrderConf(key, value, isCluster);
+            throws RemotingException, MQBrokerException, InterruptedException, MQClientException {
+        defaultMQAdminExt.createOrUpdateOrderConf(key, value, isCluster);
     }
 
     @Override
     public GroupList queryTopicConsumeByWho(String topic)
-        throws RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException,
-        InterruptedException, MQBrokerException, RemotingException, MQClientException {
-        return MQAdminInstance.threadLocalMQAdminExt().queryTopicConsumeByWho(topic);
+            throws
+            InterruptedException, MQBrokerException, RemotingException, MQClientException {
+        return defaultMQAdminExt.queryTopicConsumeByWho(topic);
     }
 
     @Override
     public boolean cleanExpiredConsumerQueue(String cluster)
-        throws RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException, MQClientException,
-        InterruptedException {
-        return MQAdminInstance.threadLocalMQAdminExt().cleanExpiredConsumerQueue(cluster);
+            throws RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException, MQClientException,
+            InterruptedException {
+        return defaultMQAdminExt.cleanExpiredConsumerQueue(cluster);
     }
 
     @Override
     public boolean cleanExpiredConsumerQueueByAddr(String addr)
-        throws RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException, MQClientException,
-        InterruptedException {
-        return MQAdminInstance.threadLocalMQAdminExt().cleanExpiredConsumerQueueByAddr(addr);
+            throws RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException, MQClientException,
+            InterruptedException {
+        return defaultMQAdminExt.cleanExpiredConsumerQueueByAddr(addr);
     }
 
     @Override
     public ConsumerRunningInfo getConsumerRunningInfo(String consumerGroup, String clientId, boolean jstack)
-        throws RemotingException, MQClientException, InterruptedException {
-        return MQAdminInstance.threadLocalMQAdminExt().getConsumerRunningInfo(consumerGroup, clientId, jstack);
+            throws RemotingException, MQClientException, InterruptedException {
+        return defaultMQAdminExt.getConsumerRunningInfo(consumerGroup, clientId, jstack);
     }
 
     @Override
     public ConsumeMessageDirectlyResult consumeMessageDirectly(String consumerGroup, String clientId,
-        String msgId) throws RemotingException, MQClientException, InterruptedException, MQBrokerException {
-        return MQAdminInstance.threadLocalMQAdminExt().consumeMessageDirectly(consumerGroup, clientId, msgId);
+                                                               String msgId) throws RemotingException, MQClientException, InterruptedException, MQBrokerException {
+        return defaultMQAdminExt.consumeMessageDirectly(consumerGroup, clientId, msgId);
     }
 
     @Override
     public List<MessageTrack> messageTrackDetail(MessageExt msg)
-        throws RemotingException, MQClientException, InterruptedException, MQBrokerException {
-        return MQAdminInstance.threadLocalMQAdminExt().messageTrackDetail(msg);
+            throws RemotingException, MQClientException, InterruptedException, MQBrokerException {
+        return defaultMQAdminExt.messageTrackDetail(msg);
     }
 
     @Override
     public void cloneGroupOffset(String srcGroup, String destGroup, String topic, boolean isOffline)
-        throws RemotingException, MQClientException, InterruptedException, MQBrokerException {
-        MQAdminInstance.threadLocalMQAdminExt().cloneGroupOffset(srcGroup, destGroup, topic, isOffline);
+            throws RemotingException, MQClientException, InterruptedException, MQBrokerException {
+        defaultMQAdminExt.cloneGroupOffset(srcGroup, destGroup, topic, isOffline);
     }
 
     @Override
     public void createTopic(String key, String newTopic, int queueNum) throws MQClientException {
-        MQAdminInstance.threadLocalMQAdminExt().createTopic(key, newTopic, queueNum);
+        defaultMQAdminExt.createTopic(key, newTopic, queueNum);
     }
 
     @Override
     public void createTopic(String key, String newTopic, int queueNum, int topicSysFlag)
-        throws MQClientException {
-        MQAdminInstance.threadLocalMQAdminExt().createTopic(key, newTopic, queueNum, topicSysFlag);
+            throws MQClientException {
+        defaultMQAdminExt.createTopic(key, newTopic, queueNum, topicSysFlag);
     }
 
     @Override
     public long searchOffset(MessageQueue mq, long timestamp) throws MQClientException {
-        return MQAdminInstance.threadLocalMQAdminExt().searchOffset(mq, timestamp);
+        return defaultMQAdminExt.searchOffset(mq, timestamp);
     }
 
     @Override
     public long maxOffset(MessageQueue mq) throws MQClientException {
-        return MQAdminInstance.threadLocalMQAdminExt().maxOffset(mq);
+        return defaultMQAdminExt.maxOffset(mq);
     }
 
     @Override
     public long minOffset(MessageQueue mq) throws MQClientException {
-        return MQAdminInstance.threadLocalMQAdminExt().minOffset(mq);
+        return defaultMQAdminExt.minOffset(mq);
     }
 
     @Override
     public long earliestMsgStoreTime(MessageQueue mq) throws MQClientException {
-        return MQAdminInstance.threadLocalMQAdminExt().earliestMsgStoreTime(mq);
+        return defaultMQAdminExt.earliestMsgStoreTime(mq);
     }
 
     @Override
     public MessageExt viewMessage(String msgId)
-        throws RemotingException, MQBrokerException, InterruptedException, MQClientException {
-        return MQAdminInstance.threadLocalMQAdminExt().viewMessage(msgId);
+            throws RemotingException, MQBrokerException, InterruptedException, MQClientException {
+        return defaultMQAdminExt.viewMessage(msgId);
     }
 
     @Override
     public QueryResult queryMessage(String topic, String key, int maxNum, long begin, long end)
-        throws MQClientException, InterruptedException {
-        return MQAdminInstance.threadLocalMQAdminExt().queryMessage(topic, key, maxNum, begin, end);
+            throws MQClientException, InterruptedException {
+        return defaultMQAdminExt.queryMessage(topic, key, maxNum, begin, end);
     }
 
     @Override
@@ -396,8 +410,8 @@ public class MQAdminExtImpl implements MQAdminExt {
 
     @Override
     public List<QueueTimeSpan> queryConsumeTimeSpan(String topic,
-        String group) throws InterruptedException, MQBrokerException, RemotingException, MQClientException {
-        return MQAdminInstance.threadLocalMQAdminExt().queryConsumeTimeSpan(topic, group);
+                                                    String group) throws InterruptedException, MQBrokerException, RemotingException, MQClientException {
+        return defaultMQAdminExt.queryConsumeTimeSpan(topic, group);
     }
 
     //MessageClientIDSetter.getNearlyTimeFromID has bug,so we subtract half a day
@@ -406,110 +420,111 @@ public class MQAdminExtImpl implements MQAdminExt {
     //https://github.com/apache/incubator-rocketmq/pull/69
     @Override
     public MessageExt viewMessage(String topic,
-        String msgId) throws RemotingException, MQBrokerException, InterruptedException, MQClientException {
+                                  String msgId) throws RemotingException, MQBrokerException, InterruptedException, MQClientException {
         logger.info("MessageClientIDSetter.getNearlyTimeFromID(msgId)={} msgId={}", MessageClientIDSetter.getNearlyTimeFromID(msgId), msgId);
         try {
             return viewMessage(msgId);
+        } catch (Exception e) {
         }
-        catch (Exception e) {
-        }
-        MQAdminImpl mqAdminImpl = MQAdminInstance.threadLocalMqClientInstance().getMQAdminImpl();
+        MQAdminImpl mqAdminImpl = mqClientInstance.getMQAdminImpl();
         QueryResult qr = Reflect.on(mqAdminImpl).call("queryMessage", topic, msgId, 32,
-            MessageClientIDSetter.getNearlyTimeFromID(msgId).getTime() - 1000 * 60 * 60 * 13L, Long.MAX_VALUE, true).get();
+                MessageClientIDSetter.getNearlyTimeFromID(msgId).getTime() - 1000 * 60 * 60 * 13L, Long.MAX_VALUE, true).get();
         if (qr != null && qr.getMessageList() != null && qr.getMessageList().size() > 0) {
             return qr.getMessageList().get(0);
-        }
-        else {
+        } else {
             return null;
         }
     }
 
     @Override
     public ConsumeMessageDirectlyResult consumeMessageDirectly(String consumerGroup, String clientId, String topic,
-        String msgId) throws RemotingException, MQClientException, InterruptedException, MQBrokerException {
-        return MQAdminInstance.threadLocalMQAdminExt().consumeMessageDirectly(consumerGroup, clientId, topic, msgId);
+                                                               String msgId) throws RemotingException, MQClientException, InterruptedException, MQBrokerException {
+        return defaultMQAdminExt.consumeMessageDirectly(consumerGroup, clientId, topic, msgId);
     }
 
     @Override
     public Properties getBrokerConfig(
-        String brokerAddr) throws RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException, UnsupportedEncodingException, InterruptedException, MQBrokerException {
-        return MQAdminInstance.threadLocalMQAdminExt().getBrokerConfig(brokerAddr);
+            String brokerAddr) throws RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException, UnsupportedEncodingException, InterruptedException, MQBrokerException {
+        return defaultMQAdminExt.getBrokerConfig(brokerAddr);
     }
 
     @Override
     public TopicList fetchTopicsByCLuster(
-        String clusterName) throws RemotingException, MQClientException, InterruptedException {
-        return MQAdminInstance.threadLocalMQAdminExt().fetchTopicsByCLuster(clusterName);
+            String clusterName) throws RemotingException, MQClientException, InterruptedException {
+        return defaultMQAdminExt.fetchTopicsByCLuster(clusterName);
     }
 
     @Override
     public boolean cleanUnusedTopic(
-        String cluster) throws RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException, MQClientException, InterruptedException {
-        return MQAdminInstance.threadLocalMQAdminExt().cleanUnusedTopic(cluster);
+            String cluster) throws RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException, MQClientException, InterruptedException {
+        return defaultMQAdminExt.cleanUnusedTopic(cluster);
     }
 
     @Override
     public boolean cleanUnusedTopicByAddr(
-        String addr) throws RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException, MQClientException, InterruptedException {
-        return MQAdminInstance.threadLocalMQAdminExt().cleanUnusedTopicByAddr(addr);
+            String addr) throws RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException, MQClientException, InterruptedException {
+        return defaultMQAdminExt.cleanUnusedTopicByAddr(addr);
     }
 
     @Override
     public BrokerStatsData viewBrokerStatsData(String brokerAddr, String statsName,
-        String statsKey) throws RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException, MQClientException, InterruptedException {
-        return MQAdminInstance.threadLocalMQAdminExt().viewBrokerStatsData(brokerAddr, statsName, statsKey);
+                                               String statsKey) throws RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException, MQClientException, InterruptedException {
+        return defaultMQAdminExt.viewBrokerStatsData(brokerAddr, statsName, statsKey);
     }
 
     @Override
     public Set<String> getClusterList(
-        String topic) throws RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException, MQClientException, InterruptedException {
-        return MQAdminInstance.threadLocalMQAdminExt().getClusterList(topic);
+            String topic) throws RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException, MQClientException, InterruptedException {
+        return defaultMQAdminExt.getClusterList(topic);
     }
 
     @Override
     public ConsumeStatsList fetchConsumeStatsInBroker(String brokerAddr, boolean isOrder,
-        long timeoutMillis) throws RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException, MQClientException, InterruptedException {
-        return MQAdminInstance.threadLocalMQAdminExt().fetchConsumeStatsInBroker(brokerAddr, isOrder, timeoutMillis);
+                                                      long timeoutMillis) throws RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException, MQClientException, InterruptedException {
+        return defaultMQAdminExt.fetchConsumeStatsInBroker(brokerAddr, isOrder, timeoutMillis);
     }
 
     @Override
     public Set<String> getTopicClusterList(
-        String topic) throws InterruptedException, MQBrokerException, MQClientException, RemotingException {
-        return MQAdminInstance.threadLocalMQAdminExt().getTopicClusterList(topic);
+            String topic) throws InterruptedException, MQBrokerException, MQClientException, RemotingException {
+        return defaultMQAdminExt.getTopicClusterList(topic);
     }
 
     @Override
     public SubscriptionGroupWrapper getAllSubscriptionGroup(String brokerAddr,
-        long timeoutMillis) throws InterruptedException, RemotingTimeoutException, RemotingSendRequestException, RemotingConnectException, MQBrokerException {
-        return MQAdminInstance.threadLocalMQAdminExt().getAllSubscriptionGroup(brokerAddr, timeoutMillis);
+                                                            long timeoutMillis) throws InterruptedException, RemotingTimeoutException, RemotingSendRequestException, RemotingConnectException, MQBrokerException {
+        return defaultMQAdminExt.getAllSubscriptionGroup(brokerAddr, timeoutMillis);
     }
 
     @Override
     public TopicConfigSerializeWrapper getAllTopicGroup(String brokerAddr,
-        long timeoutMillis) throws InterruptedException, RemotingTimeoutException, RemotingSendRequestException, RemotingConnectException, MQBrokerException {
-        return MQAdminInstance.threadLocalMQAdminExt().getAllTopicGroup(brokerAddr, timeoutMillis);
+                                                        long timeoutMillis) throws InterruptedException, RemotingTimeoutException, RemotingSendRequestException, RemotingConnectException, MQBrokerException {
+        return defaultMQAdminExt.getAllTopicGroup(brokerAddr, timeoutMillis);
     }
 
     @Override
     public void updateConsumeOffset(String brokerAddr, String consumeGroup, MessageQueue mq,
-        long offset) throws RemotingException, InterruptedException, MQBrokerException {
-        MQAdminInstance.threadLocalMQAdminExt().updateConsumeOffset(brokerAddr, consumeGroup, mq, offset);
+                                    long offset) throws RemotingException, InterruptedException, MQBrokerException {
+        defaultMQAdminExt.updateConsumeOffset(brokerAddr, consumeGroup, mq, offset);
     }
 
     // 4.0.0 added
-    @Override public void updateNameServerConfig(Properties properties,
-        List<String> list) throws InterruptedException, RemotingConnectException, UnsupportedEncodingException, RemotingSendRequestException, RemotingTimeoutException, MQClientException, MQBrokerException {
+    @Override
+    public void updateNameServerConfig(Properties properties,
+                                       List<String> list) throws InterruptedException, RemotingConnectException, UnsupportedEncodingException, RemotingSendRequestException, RemotingTimeoutException, MQClientException, MQBrokerException {
 
     }
 
-    @Override public Map<String, Properties> getNameServerConfig(
-        List<String> list) throws InterruptedException, RemotingTimeoutException, RemotingSendRequestException, RemotingConnectException, MQClientException, UnsupportedEncodingException {
+    @Override
+    public Map<String, Properties> getNameServerConfig(
+            List<String> list) throws InterruptedException, RemotingTimeoutException, RemotingSendRequestException, RemotingConnectException, MQClientException, UnsupportedEncodingException {
         return null;
     }
 
-    @Override public QueryConsumeQueueResponseBody queryConsumeQueue(String brokerAddr, String topic,
-        int queueId, long index, int count,
-        String consumerGroup) throws InterruptedException, RemotingTimeoutException, RemotingSendRequestException, RemotingConnectException, MQClientException {
+    @Override
+    public QueryConsumeQueueResponseBody queryConsumeQueue(String brokerAddr, String topic,
+                                                           int queueId, long index, int count,
+                                                           String consumerGroup) throws InterruptedException, RemotingTimeoutException, RemotingSendRequestException, RemotingConnectException, MQClientException {
         return null;
     }
 }
diff --git a/src/main/java/org/apache/rocketmq/exporter/service/client/MQAdminInstance.java b/src/main/java/org/apache/rocketmq/exporter/service/client/MQAdminInstance.java
index 6994c23..1b9b3df 100644
--- a/src/main/java/org/apache/rocketmq/exporter/service/client/MQAdminInstance.java
+++ b/src/main/java/org/apache/rocketmq/exporter/service/client/MQAdminInstance.java
@@ -16,104 +16,72 @@
  */
 package org.apache.rocketmq.exporter.service.client;
 
+import org.apache.commons.lang3.StringUtils;
 import org.apache.rocketmq.client.consumer.DefaultMQPullConsumer;
 import org.apache.rocketmq.client.exception.MQClientException;
 import org.apache.rocketmq.client.impl.MQClientAPIImpl;
 import org.apache.rocketmq.client.impl.factory.MQClientInstance;
 import org.apache.rocketmq.common.MixAll;
+import org.apache.rocketmq.exporter.config.RMQConfigure;
 import org.apache.rocketmq.remoting.RemotingClient;
 import org.apache.rocketmq.tools.admin.DefaultMQAdminExt;
 import org.apache.rocketmq.tools.admin.DefaultMQAdminExtImpl;
-import org.apache.rocketmq.tools.admin.MQAdminExt;
 import org.joor.Reflect;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.context.annotation.Bean;
+import org.springframework.stereotype.Service;
 
 import static org.apache.rocketmq.common.MixAll.TOOLS_CONSUMER_GROUP;
 
-
+@Service
 public class MQAdminInstance {
-
-    private static final ThreadLocal<DefaultMQAdminExt> MQ_ADMIN_EXT_THREAD_LOCAL = new ThreadLocal<DefaultMQAdminExt>();
-
-    private static final ThreadLocal<DefaultMQPullConsumer> MQ_PULL_CONSUMER_THREAD_LOCAL = new ThreadLocal<DefaultMQPullConsumer>();
-
-    private static final ThreadLocal<Integer> INIT_COUNTER = new ThreadLocal<Integer>();
-
-    public static MQAdminExt threadLocalMQAdminExt() {
-        DefaultMQAdminExt defaultMQAdminExt = MQ_ADMIN_EXT_THREAD_LOCAL.get();
-        if (defaultMQAdminExt == null) {
-            throw new IllegalStateException("defaultMQAdminExt should be init before you get this");
+    private final static Logger log = LoggerFactory.getLogger(MQAdminInstance.class);
+    @Autowired
+    private RMQConfigure configure;
+
+    @Bean(destroyMethod = "shutdown", name = "defaultMQAdminExt")
+    private DefaultMQAdminExt buildDefaultMQAdminExt() {
+        DefaultMQAdminExt defaultMQAdminExt = new DefaultMQAdminExt(5000L);
+        defaultMQAdminExt.setInstanceName("admin-" + System.currentTimeMillis());
+        try {
+            defaultMQAdminExt.start();
+        } catch (MQClientException ex) {
+            log.error(String.format("init default admin error, namesrv=%s", System.getProperty(MixAll.NAMESRV_ADDR_PROPERTY)), ex);
         }
         return defaultMQAdminExt;
     }
 
-
-    public static DefaultMQPullConsumer threadLocalMQPullConsumer() {
-        DefaultMQPullConsumer pullConsumer = MQ_PULL_CONSUMER_THREAD_LOCAL.get();
-        if (pullConsumer == null) {
-            throw new IllegalStateException("pullConsumer should be init before you get this");
+    @Bean(destroyMethod = "shutdown")
+    private DefaultMQPullConsumer buildPullConsumer() throws Exception {
+        String namesrvAddress = configure.getNamesrvAddr();
+        if (StringUtils.isBlank(namesrvAddress)) {
+            log.error("init default pull consumer error, namesrv is null");
+            throw new Exception("init default pull consumer error, namesrv is null", null);
         }
-        return pullConsumer;
-    }
-
-
-    public static RemotingClient threadLocalRemotingClient() {
-        MQClientInstance mqClientInstance = threadLocalMqClientInstance();
-        MQClientAPIImpl mQClientAPIImpl = Reflect.on(mqClientInstance).get("mQClientAPIImpl");
-        return Reflect.on(mQClientAPIImpl).get("remotingClient");
-    }
-
-    public static MQClientInstance threadLocalMqClientInstance() {
-        DefaultMQAdminExtImpl defaultMQAdminExtImpl = Reflect.on(MQAdminInstance.threadLocalMQAdminExt()).get("defaultMQAdminExtImpl");
-        return Reflect.on(defaultMQAdminExtImpl).get("mqClientInstance");
-    }
-
-    public static void initMQAdminInstance(long timeoutMillis) throws MQClientException {
-        Integer nowCount = INIT_COUNTER.get();
-        if (nowCount == null) {
-            DefaultMQAdminExt defaultMQAdminExt;
-            if (timeoutMillis > 0) {
-                defaultMQAdminExt = new DefaultMQAdminExt(timeoutMillis);
-            }
-            else {
-                defaultMQAdminExt = new DefaultMQAdminExt();
-            }
-            defaultMQAdminExt.setInstanceName("admin-" + Long.toString(System.currentTimeMillis()));
-            defaultMQAdminExt.start();
-            MQ_ADMIN_EXT_THREAD_LOCAL.set(defaultMQAdminExt);
-
-
-            DefaultMQPullConsumer   pullConsumer;
-            pullConsumer    =   new DefaultMQPullConsumer(TOOLS_CONSUMER_GROUP,null);
-            pullConsumer.setInstanceName("consumer-" + Long.toString(System.currentTimeMillis()));
-            pullConsumer.setNamesrvAddr(System.getProperty(MixAll.NAMESRV_ADDR_PROPERTY, System.getenv(MixAll.NAMESRV_ADDR_ENV)));
+        DefaultMQPullConsumer pullConsumer = new DefaultMQPullConsumer(TOOLS_CONSUMER_GROUP, null);
+        pullConsumer.setInstanceName("consumer-" + System.currentTimeMillis());
+        pullConsumer.setNamesrvAddr(namesrvAddress);
+        try {
             pullConsumer.start();
             pullConsumer.getDefaultMQPullConsumerImpl().getPullAPIWrapper().setConnectBrokerByUser(true);
-
-            MQ_PULL_CONSUMER_THREAD_LOCAL.set(pullConsumer);
-            INIT_COUNTER.set(1);
-        }
-        else {
-            INIT_COUNTER.set(nowCount + 1);
+        } catch (MQClientException ex) {
+            log.error(String.format("init default pull consumer error, namesrv=%s", System.getProperty(MixAll.NAMESRV_ADDR_PROPERTY)), ex);
         }
+        return pullConsumer;
+    }
 
+    @Bean(destroyMethod = "shutdown")
+    private MQClientInstance buildInstance(@Qualifier("defaultMQAdminExt") DefaultMQAdminExt defaultMQAdminExt) {
+        DefaultMQAdminExtImpl defaultMQAdminExtImpl = Reflect.on(defaultMQAdminExt).get("defaultMQAdminExtImpl");
+        return Reflect.on(defaultMQAdminExtImpl).get("mqClientInstance");
     }
 
-    public static void destroyMQAdminInstance() {
-        Integer nowCount = INIT_COUNTER.get() - 1;
-        if (nowCount > 0) {
-            INIT_COUNTER.set(nowCount);
-            return;
-        }
-        MQAdminExt mqAdminExt = MQ_ADMIN_EXT_THREAD_LOCAL.get();
-        if (mqAdminExt != null) {
-            DefaultMQPullConsumer consumer = MQ_PULL_CONSUMER_THREAD_LOCAL.get();
-            if (consumer != null) {
-                consumer.shutdown();
-                MQ_PULL_CONSUMER_THREAD_LOCAL.remove();
-            }
-            mqAdminExt.shutdown();
-            MQ_ADMIN_EXT_THREAD_LOCAL.remove();
-            INIT_COUNTER.remove();
-        }
+    @Bean
+    private RemotingClient client(MQClientInstance instance) {
+        MQClientAPIImpl mQClientAPIImpl = Reflect.on(instance).get("mQClientAPIImpl");
+        return Reflect.on(mQClientAPIImpl).get("remotingClient");
     }
 }
diff --git a/src/main/java/org/apache/rocketmq/exporter/service/impl/RMQMetricsServiceImpl.java b/src/main/java/org/apache/rocketmq/exporter/service/impl/RMQMetricsServiceImpl.java
index 5dd008b..6cf977f 100644
--- a/src/main/java/org/apache/rocketmq/exporter/service/impl/RMQMetricsServiceImpl.java
+++ b/src/main/java/org/apache/rocketmq/exporter/service/impl/RMQMetricsServiceImpl.java
@@ -19,25 +19,17 @@ package org.apache.rocketmq.exporter.service.impl;
 import io.prometheus.client.CollectorRegistry;
 import io.prometheus.client.exporter.common.TextFormat;
 import org.apache.rocketmq.exporter.collector.RMQMetricsCollector;
-import org.apache.rocketmq.exporter.service.AbstractCommonService;
 import org.apache.rocketmq.exporter.service.RMQMetricsService;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 import org.springframework.stereotype.Service;
 
 import java.io.IOException;
 import java.io.StringWriter;
 
 @Service
-public class RMQMetricsServiceImpl extends AbstractCommonService implements RMQMetricsService {
-
-    private Logger logger = LoggerFactory.getLogger(RMQMetricsServiceImpl.class);
-
-    private  CollectorRegistry registry = new CollectorRegistry();
-
+public class RMQMetricsServiceImpl implements RMQMetricsService {
+    private CollectorRegistry registry = new CollectorRegistry();
     private final RMQMetricsCollector rmqMetricsCollector;
 
-
     public RMQMetricsCollector getCollector() {
         return rmqMetricsCollector;
     }
@@ -46,8 +38,8 @@ public class RMQMetricsServiceImpl extends AbstractCommonService implements RMQM
         rmqMetricsCollector = new RMQMetricsCollector();
         rmqMetricsCollector.register(registry);
     }
-    public void Metrics(StringWriter writer) throws IOException {
+
+    public void metrics(StringWriter writer) throws IOException {
         TextFormat.write004(writer, registry.metricFamilySamples());
-        logger.info(writer.toString());
     }
 }
diff --git a/src/main/java/org/apache/rocketmq/exporter/task/MetricsCollectTask.java b/src/main/java/org/apache/rocketmq/exporter/task/MetricsCollectTask.java
index c6e6a5e..f1777ec 100644
--- a/src/main/java/org/apache/rocketmq/exporter/task/MetricsCollectTask.java
+++ b/src/main/java/org/apache/rocketmq/exporter/task/MetricsCollectTask.java
@@ -16,290 +16,483 @@
  */
 package org.apache.rocketmq.exporter.task;
 
-import com.google.common.base.Throwables;
-import org.apache.rocketmq.client.consumer.PullResult;
-import org.apache.rocketmq.client.consumer.PullStatus;
+import com.alibaba.fastjson.JSON;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.rocketmq.client.exception.MQBrokerException;
+import org.apache.rocketmq.client.exception.MQClientException;
 import org.apache.rocketmq.common.MixAll;
 import org.apache.rocketmq.common.admin.ConsumeStats;
 import org.apache.rocketmq.common.admin.OffsetWrapper;
 import org.apache.rocketmq.common.admin.TopicOffset;
 import org.apache.rocketmq.common.admin.TopicStatsTable;
 import org.apache.rocketmq.common.message.MessageQueue;
+import org.apache.rocketmq.common.protocol.ResponseCode;
 import org.apache.rocketmq.common.protocol.body.BrokerStatsData;
 import org.apache.rocketmq.common.protocol.body.ClusterInfo;
+import org.apache.rocketmq.common.protocol.body.ConsumerConnection;
 import org.apache.rocketmq.common.protocol.body.GroupList;
+import org.apache.rocketmq.common.protocol.body.KVTable;
 import org.apache.rocketmq.common.protocol.body.TopicList;
 import org.apache.rocketmq.common.protocol.route.BrokerData;
 import org.apache.rocketmq.common.protocol.route.TopicRouteData;
-import org.apache.rocketmq.exporter.aspect.admin.annotation.MultiMQAdminCmdMethod;
 import org.apache.rocketmq.exporter.config.RMQConfigure;
+import org.apache.rocketmq.exporter.model.BrokerRuntimeStats;
 import org.apache.rocketmq.exporter.service.RMQMetricsService;
-import org.apache.rocketmq.exporter.service.client.MQAdminExtImpl;
 import org.apache.rocketmq.exporter.util.Utils;
+import org.apache.rocketmq.remoting.exception.RemotingConnectException;
+import org.apache.rocketmq.remoting.exception.RemotingException;
+import org.apache.rocketmq.remoting.exception.RemotingSendRequestException;
+import org.apache.rocketmq.remoting.exception.RemotingTimeoutException;
 import org.apache.rocketmq.store.stats.BrokerStatsManager;
 import org.apache.rocketmq.tools.admin.MQAdminExt;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.scheduling.annotation.Scheduled;
 import org.springframework.stereotype.Component;
 
+import javax.annotation.PostConstruct;
 import javax.annotation.Resource;
-import java.util.Date;
-import java.util.HashMap;
 import java.util.Map;
 import java.util.Set;
 
 @Component
 public class MetricsCollectTask {
-
     @Resource
+    @Qualifier("mqAdminExtImpl")
     private MQAdminExt mqAdminExt;
     @Resource
     private RMQConfigure rmqConfigure;
-
     @Resource
-    private RMQMetricsService  metricsService;
-
+    private RMQMetricsService metricsService;
     private final static Logger log = LoggerFactory.getLogger(MetricsCollectTask.class);
 
-    @Scheduled(cron = "15 0/1 * * * ?")
-    @MultiMQAdminCmdMethod(timeoutMillis = 5000)
-    public void collectOffset() {
+    @PostConstruct
+    public void init() throws InterruptedException, RemotingConnectException, RemotingTimeoutException, RemotingSendRequestException, MQBrokerException {
+        log.info("MetricsCollectTask init starting....");
+        long start = System.currentTimeMillis();
+        ClusterInfo clusterInfo = mqAdminExt.examineBrokerClusterInfo();
+        StringBuilder infoOut = new StringBuilder();
+        for (String clusterName : clusterInfo.getClusterAddrTable().keySet()) {
+            infoOut.append(String.format("cluster name= %s, broker name = %s%n", clusterName, clusterInfo.getClusterAddrTable().get(clusterName)));
+        }
+        for (String brokerName : clusterInfo.getBrokerAddrTable().keySet()) {
+            infoOut.append(String.format("broker name = %s,master broker address= %s%n", brokerName, clusterInfo.getBrokerAddrTable().get(brokerName).getBrokerAddrs().get(MixAll.MASTER_ID)));
+        }
+        log.info(infoOut.toString());
+        log.info(String.format("MetricsCollectTask init finished....cost:%d", System.currentTimeMillis() - start));
+    }
+
+    @Scheduled(cron = "${task.collectTopicOffset.cron}")
+    public void collectTopicOffset() {
         if (!rmqConfigure.isEnableCollect()) {
             return;
         }
-        Date date = new Date();
+        log.info("topic offset collection task starting....");
+        long start = System.currentTimeMillis();
+        TopicList topicList = null;
         try {
-            TopicList topicList = mqAdminExt.fetchAllTopicList();
-            Set<String> topicSet = topicList.getTopicList();
-            for (String topic : topicSet) {
-                if (topic.startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX) || topic.startsWith(MixAll.DLQ_GROUP_TOPIC_PREFIX)) {
-                    continue;
+            topicList = mqAdminExt.fetchAllTopicList();
+        } catch (Exception ex) {
+            log.error(String.format("collectTopicOffset-exception comes getting topic list from namesrv, address is %s",
+                    JSON.toJSONString(mqAdminExt.getNameServerAddressList())));
+            return;
+        }
+        Set<String> topicSet = topicList != null ? topicList.getTopicList() : null;
+        if (topicSet == null || topicSet.isEmpty()) {
+            log.error(String.format("collectTopicOffset-the topic list is empty. the namesrv address is %s",
+                    JSON.toJSONString(mqAdminExt.getNameServerAddressList())));
+            return;
+        }
+        for (String topic : topicSet) {
+            TopicStatsTable topicStats = null;
+            try {
+                topicStats = mqAdminExt.examineTopicStats(topic);
+            } catch (Exception ex) {
+                log.error(String.format("collectTopicOffset-getting topic(%s) stats error. the namesrv address is %s",
+                        topic,
+                        JSON.toJSONString(mqAdminExt.getNameServerAddressList())));
+                continue;
+            }
+
+            Set<Map.Entry<MessageQueue, TopicOffset>> topicStatusEntries = topicStats.getOffsetTable().entrySet();
+
+            double totalMaxOffset = 0L;
+            long lastUpdateTimestamp = 0L;
+            StringBuilder sb = new StringBuilder();
+
+            for (Map.Entry<MessageQueue, TopicOffset> topicStatusEntry : topicStatusEntries) {
+                MessageQueue q = topicStatusEntry.getKey();
+                TopicOffset offset = topicStatusEntry.getValue();
+                totalMaxOffset += offset.getMaxOffset();
+                if (offset.getLastUpdateTimestamp() > lastUpdateTimestamp) {
+                    lastUpdateTimestamp = offset.getLastUpdateTimestamp();
+                }
+                sb.append(q.getBrokerName()).append(" ");
+            }
+            metricsService.getCollector().addTopicOffsetMetric("", sb.toString(), topic, lastUpdateTimestamp, totalMaxOffset);
+        }
+        log.info("topic offset collection task finished...." + (System.currentTimeMillis() - start));
+    }
+
+    @Scheduled(cron = "${task.collectConsumerOffset.cron}")
+    public void collectConsumerOffset() {
+        if (!rmqConfigure.isEnableCollect()) {
+            return;
+        }
+        log.info("consumer offset collection task starting....");
+        long start = System.currentTimeMillis();
+        TopicList topicList = null;
+        try {
+            topicList = mqAdminExt.fetchAllTopicList();
+        } catch (Exception ex) {
+            log.error(String.format("collectConsumerOffset-fetch topic list from namesrv error, the address is %s",
+                    JSON.toJSONString(mqAdminExt.getNameServerAddressList())), ex);
+            return;
+        }
+
+
+        Set<String> topicSet = topicList.getTopicList();
+        for (String topic : topicSet) {
+            GroupList groupList = null;
+
+            boolean isDLQTopic = topic.startsWith(MixAll.DLQ_GROUP_TOPIC_PREFIX);
+            if (isDLQTopic) {
+                continue;
+            }
+            try {
+                groupList = mqAdminExt.queryTopicConsumeByWho(topic);
+            } catch (Exception ex) {
+                log.warn(String.format("collectConsumerOffset-topic's consumer is empty, %s", topic));
+                continue;
+            }
+
+            if (groupList == null || groupList.getGroupList() == null || groupList.getGroupList().isEmpty()) {
+                log.warn(String.format("no any consumer for topic(%s), ignore this topic", topic));
+                continue;
+            }
+
+
+            for (String group : groupList.getGroupList()) {
+                ConsumeStats consumeStats = null;
+                ConsumerConnection onlineConsumers = null;
+                long diff = 0L, totalConsumerOffset = 0L, totalBrokerOffset = 0L;
+                int countOfOnlineConsumers = 0;
+
+                double consumeTPS = 0F;
+                try {
+                    onlineConsumers = mqAdminExt.examineConsumerConnectionInfo(group);
+                } catch (InterruptedException | RemotingException ex) {
+                    log.error(String.format("get topic's(%s) online consumers(%s) exception", topic, group), ex);
+                } catch (MQClientException ex) {
+                    handleTopicNotExistException(ex.getResponseCode(), ex, topic, group);
+                } catch (MQBrokerException ex) {
+                    handleTopicNotExistException(ex.getResponseCode(), ex, topic, group);
                 }
-                String clusterName = null;
-                ClusterInfo clusterInfo = mqAdminExt.examineBrokerClusterInfo();
-                Set<Map.Entry<String, BrokerData>> clusterEntries = clusterInfo.getBrokerAddrTable().entrySet();
-                for (Map.Entry<String, BrokerData> clusterEntry : clusterEntries) {
-                    clusterName  = clusterEntry.getValue().getCluster();
-                    break;
+                if (onlineConsumers == null || onlineConsumers.getConnectionSet() == null || onlineConsumers.getConnectionSet().isEmpty()) {
+                    log.warn(String.format("no any consumer online. topic=%s, consumer group=%s. ignore this", topic, group));
+                    countOfOnlineConsumers = 0;
+                } else {
+                    countOfOnlineConsumers = onlineConsumers.getConnectionSet().size();
                 }
-                if (clusterName != null) {
-                    HashMap<String,Long>    brokerOffsetMap = new HashMap<>();
-                    TopicStatsTable topicStatus = mqAdminExt.examineTopicStats(topic);
-                    Set<Map.Entry<MessageQueue, TopicOffset>> topicStatusEntries = topicStatus.getOffsetTable().entrySet();
-                    for (Map.Entry<MessageQueue, TopicOffset> topicStatusEntry : topicStatusEntries) {
-                        MessageQueue q      =   topicStatusEntry.getKey();
-                        TopicOffset offset  =   topicStatusEntry.getValue();
-                        if  (brokerOffsetMap.containsKey(q.getBrokerName())) {
-                            brokerOffsetMap.put(q.getBrokerName(),brokerOffsetMap.get(q.getBrokerName()) + offset.getMaxOffset());
-                        }
-                        else {
-                            brokerOffsetMap.put(q.getBrokerName(),offset.getMaxOffset());
-                        }
-                    }
-                    Set<Map.Entry<String, Long>> brokerOffsetEntries = brokerOffsetMap.entrySet();
-                    for (Map.Entry<String, Long> brokerOffsetEntry : brokerOffsetEntries) {
-                        metricsService.getCollector().AddTopicOffsetMetric(clusterName,brokerOffsetEntry.getKey(), topic, brokerOffsetEntry.getValue());
-                    }
+                try {
+                    consumeStats = mqAdminExt.examineConsumeStats(group, topic);
+                } catch (InterruptedException | RemotingException ex) {
+                    log.error(String.format("get topic's(%s) consumer-stats(%s) exception", topic, group), ex);
+                } catch (MQClientException ex) {
+                    handleTopicNotExistException(ex.getResponseCode(), ex, topic, group);
+                } catch (MQBrokerException ex) {
+                    handleTopicNotExistException(ex.getResponseCode(), ex, topic, group);
+                }
+                if (consumeStats == null || consumeStats.getOffsetTable() == null || consumeStats.getOffsetTable().isEmpty()) {
+                    log.warn(String.format("no any offset for consumer(%s), topic(%s), ignore this", group, topic));
+                    continue;
+                }
+                {
+                    diff = consumeStats.computeTotalDiff();
+                    consumeTPS = consumeStats.getConsumeTps();
+                    metricsService.getCollector().addGroupDiffMetric(String.valueOf(countOfOnlineConsumers), group, topic, diff);
+                    metricsService.getCollector().addGroupConsumeTPSMetric(topic, group, consumeTPS);
                 }
+                Set<Map.Entry<MessageQueue, OffsetWrapper>> consumeStatusEntries = consumeStats.getOffsetTable().entrySet();
+                for (Map.Entry<MessageQueue, OffsetWrapper> consumeStatusEntry : consumeStatusEntries) {
+                    MessageQueue q = consumeStatusEntry.getKey();
+                    OffsetWrapper offset = consumeStatusEntry.getValue();
 
-                HashMap<String,Long>    consumeOffsetMap = new HashMap<>();
-                GroupList groupList = mqAdminExt.queryTopicConsumeByWho(topic);
-                if (groupList != null && !groupList.getGroupList().isEmpty()) {
-                    for (String group : groupList.getGroupList()) {
-                        try {
-                            ConsumeStats consumeStatus = mqAdminExt.examineConsumeStats(group, topic);
-                            Set<Map.Entry<MessageQueue, OffsetWrapper>> consumeStatusEntries = consumeStatus.getOffsetTable().entrySet();
-                            for (Map.Entry<MessageQueue, OffsetWrapper> consumeStatusEntry : consumeStatusEntries) {
-                                MessageQueue q = consumeStatusEntry.getKey();
-                                OffsetWrapper offset = consumeStatusEntry.getValue();
-                                if (consumeOffsetMap.containsKey(q.getBrokerName())) {
-                                    consumeOffsetMap.put(q.getBrokerName(), consumeOffsetMap.get(q.getBrokerName()) + offset.getConsumerOffset());
-                                } else {
-                                    consumeOffsetMap.put(q.getBrokerName(), offset.getConsumerOffset());
-                                }
-                            }
-                        } catch (Exception e) {
-                            log.info("ignore this consumer", e.getMessage());
-                        }
-                        Set<Map.Entry<String, Long>> consumeOffsetEntries = consumeOffsetMap.entrySet();
-                        for (Map.Entry<String, Long> consumeOffsetEntry : consumeOffsetEntries) {
-                            metricsService.getCollector().AddGroupOffsetMetric(clusterName,consumeOffsetEntry.getKey(), topic, group, consumeOffsetEntry.getValue());
-                        }
-                        consumeOffsetMap.clear();
-                    }
+                    //topic + consumer group 生产offset
+                    totalBrokerOffset += totalBrokerOffset + offset.getBrokerOffset();
+                    //topic + consumer group 消费offset
+                    totalConsumerOffset += offset.getConsumerOffset();
                 }
+                metricsService.getCollector().addGroupBrokerTotalOffsetMetric(topic, group, totalBrokerOffset);
+                metricsService.getCollector().addGroupConsumerTotalOffsetMetric(topic, group, totalBrokerOffset);
             }
-        } catch (Exception e) {
-            log.info("error is " + e.getMessage());
         }
+        log.info("consumer offset collection task finished...." + (System.currentTimeMillis() - start));
     }
 
-    @Scheduled(cron = "15 0/1 * * * ?")
-    @MultiMQAdminCmdMethod(timeoutMillis = 5000)
-    public void collectTopic() {
+    @Scheduled(cron = "${task.collectBrokerStatsTopic.cron}")
+    public void collectBrokerStatsTopic() {
         if (!rmqConfigure.isEnableCollect()) {
             return;
         }
-        Date date = new Date();
+        log.info("broker topic stats collection task starting....");
+        long start = System.currentTimeMillis();
+        Set<String> topicSet = null;
         try {
             TopicList topicList = mqAdminExt.fetchAllTopicList();
-            Set<String> topicSet = topicList.getTopicList();
-            for (String topic : topicSet) {
-                if (topic.startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX) || topic.startsWith(MixAll.DLQ_GROUP_TOPIC_PREFIX)) {
-                    continue;
+            topicSet = topicList.getTopicList();
+        } catch (Exception ex) {
+            log.error(String.format("collectBrokerStatsTopic-fetch topic list from namesrv error, the address is %s",
+                    JSON.toJSONString(mqAdminExt.getNameServerAddressList())), ex);
+            return;
+        }
+        if (topicSet == null || topicSet.isEmpty()) {
+            return;
+        }
+        ClusterInfo clusterInfo = null;
+        try {
+            clusterInfo = mqAdminExt.examineBrokerClusterInfo();
+        } catch (Exception ex) {
+            log.error(String.format("collectBrokerStatsTopic-fetch cluster info exception, the address is %s",
+                    JSON.toJSONString(mqAdminExt.getNameServerAddressList())), ex);
+            return;
+        }
+
+        for (String topic : topicSet) {
+            if (topic.startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX) || topic.startsWith(MixAll.DLQ_GROUP_TOPIC_PREFIX)) {
+                continue;
+            }
+            TopicRouteData topicRouteData = null;
+
+            try {
+                topicRouteData = mqAdminExt.examineTopicRouteInfo(topic);
+            } catch (Exception ex) {
+                log.error(String.format("fetch topic route error. ignore %s", topic), ex);
+                continue;
+            }
+            for (BrokerData bd : topicRouteData.getBrokerDatas()) {
+                String masterAddr = bd.getBrokerAddrs().get(MixAll.MASTER_ID);
+                if (!StringUtils.isBlank(masterAddr)) {
+                    BrokerStatsData bsd = null;
+                    try {
+                        //topic发了多少条消息
+                        bsd = mqAdminExt.viewBrokerStatsData(masterAddr, BrokerStatsManager.TOPIC_PUT_NUMS, topic);
+                        String brokerIP = clusterInfo.getBrokerAddrTable().get(bd.getBrokerName()).getBrokerAddrs().get(MixAll.MASTER_ID);
+                        metricsService.getCollector().addTopicPutNumsMetric(
+                                bd.getCluster(),
+                                bd.getBrokerName(),
+                                brokerIP,
+                                "",
+                                topic,
+                                Utils.getFixedDouble(bsd.getStatsMinute().getTps())
+                        );
+                    } catch (MQClientException ex) {
+                        if (ex.getResponseCode() == ResponseCode.SYSTEM_ERROR) {
+                            log.error(String.format("TOPIC_PUT_NUMS-error, topic=%s, master broker=%s, %s", topic, masterAddr, ex.getErrorMessage()));
+                        } else {
+                            log.error(String.format("TOPIC_PUT_NUMS-error, topic=%s, master broker=%s", topic, masterAddr), ex);
+                        }
+                    } catch (RemotingTimeoutException | InterruptedException | RemotingSendRequestException | RemotingConnectException ex1) {
+                        log.error(String.format("TOPIC_PUT_NUMS-error, topic=%s, master broker=%s", topic, masterAddr), ex1);
+                    }
+                    try {
+                        //topic总共发了多少字节
+                        bsd = mqAdminExt.viewBrokerStatsData(masterAddr, BrokerStatsManager.TOPIC_PUT_SIZE, topic);
+                        String brokerIP = clusterInfo.getBrokerAddrTable().get(bd.getBrokerName()).getBrokerAddrs().get(MixAll.MASTER_ID);
+                        metricsService.getCollector().addTopicPutSizeMetric(
+                                bd.getCluster(),
+                                bd.getBrokerName(),
+                                brokerIP,
+                                "",
+                                topic,
+                                Utils.getFixedDouble(bsd.getStatsMinute().getTps())
+                        );
+                    } catch (MQClientException ex) {
+                        if (ex.getResponseCode() == ResponseCode.SYSTEM_ERROR) {
+                            log.error(String.format("TOPIC_PUT_SIZE-error, topic=%s, master broker=%s, %s", topic, masterAddr, ex.getErrorMessage()));
+                        } else {
+                            log.error(String.format("TOPIC_PUT_SIZE-error, topic=%s, master broker=%s", topic, masterAddr), ex);
+                        }
+                    } catch (InterruptedException | RemotingConnectException | RemotingTimeoutException | RemotingSendRequestException ex) {
+                        log.error(String.format("TOPIC_PUT_SIZE-error, topic=%s, master broker=%s", topic, masterAddr), ex);
+                    }
                 }
-                TopicRouteData topicRouteData = mqAdminExt.examineTopicRouteInfo(topic);
-                GroupList groupList = mqAdminExt.queryTopicConsumeByWho(topic);
+            }
 
+            GroupList groupList = null;
+            try {
+                groupList = mqAdminExt.queryTopicConsumeByWho(topic);
+            } catch (Exception ex) {
+                log.error(String.format("collectBrokerStatsTopic-fetch consumers for topic(%s) error, ignore this topic", topic), ex);
+                return;
+            }
+            if (groupList.getGroupList() == null || groupList.getGroupList().isEmpty()) {
+                log.warn(String.format("collectBrokerStatsTopic-topic's consumer is empty, %s", topic));
+                return;
+            }
+            for (String group : groupList.getGroupList())
                 for (BrokerData bd : topicRouteData.getBrokerDatas()) {
                     String masterAddr = bd.getBrokerAddrs().get(MixAll.MASTER_ID);
                     if (masterAddr != null) {
+                        String statsKey = String.format("%s@%s", topic, group);
+                        BrokerStatsData bsd = null;
                         try {
-                            BrokerStatsData bsd = null;
-                            try {
-                                bsd = mqAdminExt.viewBrokerStatsData(masterAddr, BrokerStatsManager.TOPIC_PUT_NUMS, topic);
-                                metricsService.getCollector().AddTopicPutNumsMetric(bd.getCluster(), bd.getBrokerName(), topic, Utils.getFixedDouble(bsd.getStatsMinute().getTps()));
-                            }
-                            catch (Exception e) {
-                                log.info("error is " + e.getMessage());
-                            }
-                            try {
-                                bsd = mqAdminExt.viewBrokerStatsData(masterAddr, BrokerStatsManager.TOPIC_PUT_SIZE, topic);
-                                metricsService.getCollector().AddTopicPutSizeMetric(bd.getCluster(), bd.getBrokerName(), topic, Utils.getFixedDouble(bsd.getStatsMinute().getTps()));
+                            //消费者消费了多少条消息
+                            bsd = mqAdminExt.viewBrokerStatsData(masterAddr, BrokerStatsManager.GROUP_GET_NUMS, statsKey);
+                            metricsService.getCollector().addGroupGetNumsMetric(
+                                    topic,
+                                    group,
+                                    Utils.getFixedDouble(bsd.getStatsMinute().getTps()));
+                        } catch (MQClientException ex) {
+                            if (ex.getResponseCode() == ResponseCode.SYSTEM_ERROR) {
+                                log.error(String.format("GROUP_GET_NUMS-error, topic=%s, group=%s,master broker=%s, %s", topic, group, masterAddr, ex.getErrorMessage()));
+                            } else {
+                                log.error(String.format("GROUP_GET_NUMS-error, topic=%s, group=%s,master broker=%s", topic, group, masterAddr), ex);
                             }
-                            catch (Exception e) {
-                                log.info("error is " + e.getMessage());
+                        } catch (InterruptedException | RemotingConnectException | RemotingTimeoutException | RemotingSendRequestException ex) {
+                            log.error(String.format("GROUP_GET_NUMS-error, topic=%s, group=%s,master broker=%s", topic, group, masterAddr), ex);
+                        }
+                        try {
+                            //消费者消费了多少字节
+                            bsd = mqAdminExt.viewBrokerStatsData(masterAddr, BrokerStatsManager.GROUP_GET_SIZE, statsKey);
+                            metricsService.getCollector().addGroupGetSizeMetric(
+                                    topic,
+                                    group,
+                                    Utils.getFixedDouble(bsd.getStatsMinute().getTps()));
+                        } catch (MQClientException ex) {
+                            if (ex.getResponseCode() == ResponseCode.SYSTEM_ERROR) {
+                                log.error(String.format("GROUP_GET_SIZE-error, topic=%s, group=%s, master broker=%s, %s", topic, group, masterAddr, ex.getErrorMessage()));
+                            } else {
+                                log.error(String.format("GROUP_GET_SIZE-error, topic=%s, group=%s, master broker=%s", topic, group, masterAddr), ex);
                             }
-                        } catch (Exception e) {
-                            log.info("error is " + e.getMessage());
+                        } catch (InterruptedException | RemotingConnectException | RemotingTimeoutException | RemotingSendRequestException ex) {
+                            log.error(String.format("GROUP_GET_SIZE-error, topic=%s, group=%s, master broker=%s", topic, group, masterAddr), ex);
                         }
-                    }
-                }
-                if (groupList != null && !groupList.getGroupList().isEmpty()) {
-                    for (String group : groupList.getGroupList()) {
-                        for (BrokerData bd : topicRouteData.getBrokerDatas()) {
-                            String masterAddr = bd.getBrokerAddrs().get(MixAll.MASTER_ID);
-                            if (masterAddr != null) {
-                                try {
-                                    String statsKey = String.format("%s@%s", topic, group);
-                                    BrokerStatsData bsd = null;
-                                    try {
-                                        bsd = mqAdminExt.viewBrokerStatsData(masterAddr, BrokerStatsManager.GROUP_GET_NUMS, statsKey);
-                                        metricsService.getCollector().AddGroupGetNumsMetric(bd.getCluster(), bd.getBrokerName(), topic, group, Utils.getFixedDouble(bsd.getStatsMinute().getTps()));
-                                    } catch (Exception e) {
-                                        log.info("error is " + e.getMessage());
-                                    }
-                                    try {
-                                        bsd = mqAdminExt.viewBrokerStatsData(masterAddr, BrokerStatsManager.GROUP_GET_SIZE, statsKey);
-                                        metricsService.getCollector().AddGroupGetSizeMetric(bd.getCluster(), bd.getBrokerName(), topic, group, Utils.getFixedDouble(bsd.getStatsMinute().getTps()));
-                                    } catch (Exception e) {
-                                        log.info("error is " + e.getMessage());
-                                    }
-                                    try {
-
-                                        bsd = mqAdminExt.viewBrokerStatsData(masterAddr, BrokerStatsManager.SNDBCK_PUT_NUMS, statsKey);
-                                        metricsService.getCollector().AddsendBackNumsMetric(bd.getCluster(), bd.getBrokerName(), topic, group, Utils.getFixedDouble(bsd.getStatsMinute().getTps()));
-                                    } catch (Exception e) {
-                                        log.info("error is " + e.getMessage());
-                                    }
-                                    try {
-                                        collectLatencyMetrcisInner(topic, group, masterAddr, bd);
-                                    } catch (Exception e) {
-                                        log.info("error is " + e.getMessage());
-                                    }
-                                } catch (Exception e) {
-                                    log.info("error is " + e.getMessage());
-                                }
+                        try {
+                            //消费者重新消费topic的次数
+                            bsd = mqAdminExt.viewBrokerStatsData(masterAddr, BrokerStatsManager.SNDBCK_PUT_NUMS, statsKey);
+                            metricsService.getCollector().addSendBackNumsMetric(
+                                    topic,
+                                    group,
+                                    Utils.getFixedDouble(bsd.getStatsMinute().getTps()));
+                        } catch (MQClientException ex) {
+                            if (ex.getResponseCode() == ResponseCode.SYSTEM_ERROR) {
+                                log.error(String.format("SNDBCK_PUT_NUMS-error, topic=%s, group=%s, master broker=%s, %s", topic, group, masterAddr, ex.getErrorMessage()));
+                            } else {
+                                log.error(String.format("SNDBCK_PUT_NUMS-error, topic=%s, group=%s, master broker=%s", topic, group, masterAddr), ex);
                             }
+                        } catch (InterruptedException | RemotingConnectException | RemotingTimeoutException | RemotingSendRequestException ex) {
+                            log.error(String.format("SNDBCK_PUT_NUMS-error, topic=%s, group=%s, master broker=%s", topic, group, masterAddr), ex);
                         }
                     }
                 }
-            }
-        }
-        catch (Exception err) {
-            throw Throwables.propagate(err);
         }
+        log.info("broker topic stats collection task finished...." + (System.currentTimeMillis() - start));
     }
 
-    @Scheduled(cron = "15 0/1 * * * ?")
-    @MultiMQAdminCmdMethod(timeoutMillis = 5000)
-    public void collectBroker() {
+    @Scheduled(cron = "${task.collectBrokerStats.cron}")
+    public void collectBrokerStats() {
         if (!rmqConfigure.isEnableCollect()) {
             return;
         }
+        log.info("broker stats collection task starting....");
+        long start = System.currentTimeMillis();
+        ClusterInfo clusterInfo = null;
         try {
-            Date date = new Date();
-            ClusterInfo clusterInfo = mqAdminExt.examineBrokerClusterInfo();
-            Set<Map.Entry<String, BrokerData>> clusterEntries = clusterInfo.getBrokerAddrTable().entrySet();
-            for (Map.Entry<String, BrokerData> clusterEntry : clusterEntries) {
-                String masterAddr = clusterEntry.getValue().getBrokerAddrs().get(MixAll.MASTER_ID);
-                if (masterAddr != null) {
-                    try {
-                        BrokerStatsData bsd = null;
-                        try {
-                            bsd = mqAdminExt.viewBrokerStatsData(masterAddr, BrokerStatsManager.BROKER_PUT_NUMS,clusterEntry.getValue().getCluster());
-                            metricsService.getCollector().AddBrokerPutNumsMetric(clusterEntry.getValue().getCluster(), clusterEntry.getValue().getBrokerName(), Utils.getFixedDouble(bsd.getStatsMinute().getTps()));
-                        }
-                        catch (Exception e) {
-                            log.info("error is " + e.getMessage());
-                        }
-                        try {
-                            bsd = mqAdminExt.viewBrokerStatsData(masterAddr, BrokerStatsManager.BROKER_GET_NUMS, clusterEntry.getValue().getCluster());
-                            metricsService.getCollector().AddBrokerGetNumsMetric(clusterEntry.getValue().getCluster(), clusterEntry.getValue().getBrokerName(), Utils.getFixedDouble(bsd.getStatsMinute().getTps()));
-                        }
-                        catch (Exception e) {
-                            log.info("error is " + e.getMessage());
-                        }
-                    } catch (Exception e) {
-                        log.info("error is " + e.getMessage());
-                    }
-                }
-            }
-        }
-        catch (Exception err) {
-            throw Throwables.propagate(err);
+            clusterInfo = mqAdminExt.examineBrokerClusterInfo();
+        } catch (Exception ex) {
+            log.error(String.format("collectBrokerStats-get cluster info from namesrv error. address is %s", JSON.toJSONString(mqAdminExt.getNameServerAddressList())), ex);
+            return;
         }
-    }
-    private void collectLatencyMetrcisInner(String topic,String group,String masterAddr, BrokerData bd) throws Exception {
-        long maxLagTime = 0;
-        String statsKey;
-        BrokerStatsData bsd = null;
-        ConsumeStats consumeStatus = mqAdminExt.examineConsumeStats(group, topic);
-        Set<Map.Entry<MessageQueue, OffsetWrapper>> consumeStatusEntries = consumeStatus.getOffsetTable().entrySet();
-        for (Map.Entry<MessageQueue, OffsetWrapper> consumeStatusEntry : consumeStatusEntries) {
-            MessageQueue q = consumeStatusEntry.getKey();
-            OffsetWrapper offset = consumeStatusEntry.getValue();
-            int queueId = q.getQueueId();
-            statsKey = String.format("%d@%s@%s", queueId, topic, group);
+
+        Set<Map.Entry<String, BrokerData>> clusterEntries = clusterInfo.getBrokerAddrTable().entrySet();
+        for (Map.Entry<String, BrokerData> clusterEntry : clusterEntries) {
+            String masterAddr = clusterEntry.getValue().getBrokerAddrs().get(MixAll.MASTER_ID);
+            if (StringUtils.isBlank(masterAddr)) {
+                continue;
+            }
+            BrokerStatsData bsd = null;
             try {
-                bsd = mqAdminExt.viewBrokerStatsData(masterAddr, BrokerStatsManager.GROUP_GET_LATENCY, statsKey);
-                metricsService.getCollector().AddGroupGetLatencyMetric(bd.getCluster(), bd.getBrokerName(), topic, group, String.format("%d", queueId), Utils.getFixedDouble(bsd.getStatsMinute().getTps()));
-            } catch (Exception e) {
-                log.info("error is " + e.getMessage());
+                bsd = mqAdminExt.viewBrokerStatsData(masterAddr, BrokerStatsManager.BROKER_PUT_NUMS, clusterEntry.getValue().getCluster());
+                String brokerIP = clusterEntry.getValue().getBrokerAddrs().get(MixAll.MASTER_ID);
+                metricsService.getCollector().addBrokerPutNumsMetric(
+                        clusterEntry.getValue().getCluster(),
+                        brokerIP,
+                        "",
+                        Utils.getFixedDouble(bsd.getStatsMinute().getTps()));
+            } catch (Exception ex) {
+                log.error(String.format("BROKER_PUT_NUMS-error, master broker=%s", masterAddr), ex);
             }
-            MQAdminExtImpl mqAdminImpl = (MQAdminExtImpl) mqAdminExt;
-            PullResult consumePullResult = mqAdminImpl.queryMsgByOffset(q, offset.getConsumerOffset());
-            long lagTime = 0;
-            if (consumePullResult != null && consumePullResult.getPullStatus() == PullStatus.FOUND) {
-                lagTime = System.currentTimeMillis() - consumePullResult.getMsgFoundList().get(0).getStoreTimestamp();
-                if (offset.getBrokerOffset() == offset.getConsumerOffset()) {
-                    lagTime = 0;
-                }
-            } else if (consumePullResult.getPullStatus() == PullStatus.NO_MATCHED_MSG) {
-                lagTime = 0;
-            } else if (consumePullResult.getPullStatus() == PullStatus.OFFSET_ILLEGAL) {
-                PullResult pullResult = mqAdminImpl.queryMsgByOffset(q, consumePullResult.getMinOffset());
-                if (pullResult != null && pullResult.getPullStatus() == PullStatus.FOUND) {
-                    lagTime = System.currentTimeMillis() - consumePullResult.getMsgFoundList().get(0).getStoreTimestamp();
+            try {
+                bsd = mqAdminExt.viewBrokerStatsData(masterAddr, BrokerStatsManager.BROKER_GET_NUMS, clusterEntry.getValue().getCluster());
+                String brokerIP = clusterEntry.getValue().getBrokerAddrs().get(MixAll.MASTER_ID);
+                metricsService.getCollector().addBrokerGetNumsMetric(
+                        clusterEntry.getValue().getCluster(),
+                        brokerIP,
+                        "",
+                        Utils.getFixedDouble(bsd.getStatsMinute().getTps()));
+            } catch (Exception ex) {
+                log.error(String.format("BROKER_GET_NUMS-error, master broker=%s", masterAddr), ex);
+            }
+        }
+        log.info("broker stats collection task finished...." + (System.currentTimeMillis() - start));
+    }
+
+    @Scheduled(cron = "${task.collectBrokerRuntimeStats.cron}")
+    public void collectBrokerRuntimeStats() {
+        if (!rmqConfigure.isEnableCollect()) {
+            return;
+        }
+        log.info("broker runtime stats collection task starting....");
+        long start = System.currentTimeMillis();
+        ClusterInfo clusterInfo = null;
+        try {
+            clusterInfo = mqAdminExt.examineBrokerClusterInfo();
+        } catch (Exception ex) {
+            log.error(String.format("collectBrokerRuntimeStats-get cluster info from namesrv error. address is %s", JSON.toJSONString(mqAdminExt.getNameServerAddressList())), ex);
+            return;
+        }
+
+        Set<Map.Entry<String, BrokerData>> clusterEntries = clusterInfo.getBrokerAddrTable().entrySet();
+        for (Map.Entry<String, BrokerData> clusterEntry : clusterEntries) {
+            String masterAddr = clusterEntry.getValue().getBrokerAddrs().get(MixAll.MASTER_ID);
+            String clusterName = clusterEntry.getValue().getCluster();
+
+            KVTable kvTable = null;
+            if (!StringUtils.isBlank(masterAddr)) {
+                try {
+                    kvTable = mqAdminExt.fetchBrokerRuntimeStats(masterAddr);
+                } catch (RemotingConnectException | RemotingSendRequestException | RemotingTimeoutException | InterruptedException ex) {
+                    log.error(String.format("collectBrokerRuntimeStats-get fetch broker runtime stats error, address=%s", masterAddr), ex);
+                } catch (MQBrokerException ex) {
+                    if (ex.getResponseCode() == ResponseCode.SYSTEM_ERROR) {
+                        log.error(String.format("collectBrokerRuntimeStats-get fetch broker runtime stats error, address=%s, error=%s", masterAddr, ex.getErrorMessage()));
+                    } else {
+                        log.error(String.format("collectBrokerRuntimeStats-get fetch broker runtime stats error, address=%s", masterAddr), ex);
+                    }
                 }
-            } else {
-                lagTime = 0;
             }
-            if (lagTime > maxLagTime) {
-                maxLagTime = lagTime;
+            if (kvTable == null || kvTable.getTable() == null || kvTable.getTable().isEmpty()) {
+                continue;
             }
+            try {
+                BrokerRuntimeStats brokerRuntimeStats = new BrokerRuntimeStats(kvTable);
+                metricsService.getCollector().addBrokerRuntimeStatsMetric(brokerRuntimeStats, clusterName, masterAddr, "");
+            } catch (Exception ex) {
+                log.error(String.format("collectBrokerRuntimeStats-parse or report broker runtime stats error, %s", JSON.toJSONString(kvTable)), ex);
+            }
+
+        }
+
+        log.info("broker runtime stats collection task finished...." + (System.currentTimeMillis() - start));
+    }
+
+    private void handleTopicNotExistException(int responseCode, Exception ex, String topic, String group) {
+        if (responseCode == ResponseCode.TOPIC_NOT_EXIST || responseCode == ResponseCode.CONSUMER_NOT_ONLINE) {
+            log.error(String.format("get topic's(%s) consumer-stats(%s) exception, detail: %s", topic, group, ex.getMessage()));
+        } else {
+            log.error(String.format("get topic's(%s) consumer-stats(%s) exception", topic, group), ex);
         }
-        metricsService.getCollector().AddGroupGetLatencyByStoreTimeMetric(bd.getCluster(), bd.getBrokerName(), topic, group, maxLagTime);
     }
 }
diff --git a/src/main/java/org/apache/rocketmq/exporter/util/JsonUtil.java b/src/main/java/org/apache/rocketmq/exporter/util/JsonUtil.java
index 29317f9..1cb147c 100644
--- a/src/main/java/org/apache/rocketmq/exporter/util/JsonUtil.java
+++ b/src/main/java/org/apache/rocketmq/exporter/util/JsonUtil.java
@@ -55,8 +55,7 @@ public class JsonUtil {
     public static void writeValue(Writer writer, Object obj) {
         try {
             objectMapper.writeValue(writer, obj);
-        }
-        catch (IOException e) {
+        } catch (IOException e) {
             Throwables.propagateIfPossible(e);
         }
     }
@@ -67,9 +66,8 @@ public class JsonUtil {
         }
 
         try {
-            return src instanceof String ? (String)src : objectMapper.writeValueAsString(src);
-        }
-        catch (Exception e) {
+            return src instanceof String ? (String) src : objectMapper.writeValueAsString(src);
+        } catch (Exception e) {
             logger.error("Parse Object to String error src=" + src, e);
             return null;
         }
@@ -81,9 +79,8 @@ public class JsonUtil {
         }
 
         try {
-            return src instanceof byte[] ? (byte[])src : objectMapper.writeValueAsBytes(src);
-        }
-        catch (Exception e) {
+            return src instanceof byte[] ? (byte[]) src : objectMapper.writeValueAsBytes(src);
+        } catch (Exception e) {
             logger.error("Parse Object to byte[] error", e);
             return null;
         }
@@ -95,9 +92,8 @@ public class JsonUtil {
         }
         str = escapesSpecialChar(str);
         try {
-            return clazz.equals(String.class) ? (T)str : objectMapper.readValue(str, clazz);
-        }
-        catch (Exception e) {
+            return clazz.equals(String.class) ? (T) str : objectMapper.readValue(str, clazz);
+        } catch (Exception e) {
             logger.error("Parse String to Object error\nString: {}\nClass<T>: {}\nError: {}", str, clazz.getName(), e);
             return null;
         }
@@ -108,9 +104,8 @@ public class JsonUtil {
             return null;
         }
         try {
-            return clazz.equals(byte[].class) ? (T)bytes : objectMapper.readValue(bytes, clazz);
-        }
-        catch (Exception e) {
+            return clazz.equals(byte[].class) ? (T) bytes : objectMapper.readValue(bytes, clazz);
+        } catch (Exception e) {
             logger.error("Parse byte[] to Object error\nbyte[]: {}\nClass<T>: {}\nError: {}", bytes, clazz.getName(), e);
             return null;
         }
@@ -122,11 +117,10 @@ public class JsonUtil {
         }
         str = escapesSpecialChar(str);
         try {
-            return (T)(typeReference.getType().equals(String.class) ? str : objectMapper.readValue(str, typeReference));
-        }
-        catch (Exception e) {
+            return (T) (typeReference.getType().equals(String.class) ? str : objectMapper.readValue(str, typeReference));
+        } catch (Exception e) {
             logger.error("Parse String to Object error\nString: {}\nTypeReference<T>: {}\nError: {}", str,
-                typeReference.getType(), e);
+                    typeReference.getType(), e);
             return null;
         }
     }
@@ -136,12 +130,11 @@ public class JsonUtil {
             return null;
         }
         try {
-            return (T)(typeReference.getType().equals(byte[].class) ? bytes : objectMapper.readValue(bytes,
-                typeReference));
-        }
-        catch (Exception e) {
+            return (T) (typeReference.getType().equals(byte[].class) ? bytes : objectMapper.readValue(bytes,
+                    typeReference));
+        } catch (Exception e) {
             logger.error("Parse byte[] to Object error\nbyte[]: {}\nTypeReference<T>: {}\nError: {}", bytes,
-                typeReference.getType(), e);
+                    typeReference.getType(), e);
             return null;
         }
     }
diff --git a/src/main/resources/application.properties b/src/main/resources/application.properties
deleted file mode 100644
index 083738e..0000000
--- a/src/main/resources/application.properties
+++ /dev/null
@@ -1,14 +0,0 @@
-server.port=5557
-
-spring.application.name=rocketmq-exporter
-spring.http.encoding.charset=UTF-8
-spring.http.encoding.enabled=true
-spring.http.encoding.force=true
-logging.config=classpath:logback.xml
-#if this value is empty,use env value rocketmq.config.namesrvAddr  NAMESRV_ADDR
-rocketmq.config.namesrvAddr=127.0.0.1:9876
-
-
-rocketmq.config.enableCollect=true
-rocketmq.config.webTelemetryPath=/metrics
-rocketmq.config.rocketmqVersion=4_3_2
\ No newline at end of file
diff --git a/src/main/resources/application.yml b/src/main/resources/application.yml
new file mode 100644
index 0000000..8ff9e07
--- /dev/null
+++ b/src/main/resources/application.yml
@@ -0,0 +1,32 @@
+server:
+  port: 5557
+spring:
+  application:
+    name: rocketmq-exporter
+  http:
+    encoding:
+      charset: UTF-8
+      enabled: true
+      force: true
+logging:
+  config: classpath:logback.xml
+
+rocketmq:
+  config:
+    webTelemetryPath: /metrics
+    rocketmqVersion: 4_2_0
+    namesrvAddr: 127.0.0.1:9876 #
+    enableCollect: true
+
+task:
+  count: 5
+  collectTopicOffset:
+    cron: 15 0/1 * * * ?
+  collectConsumerOffset:
+    cron: 15 0/1 * * * ?
+  collectBrokerStatsTopic:
+    cron: 15 0/1 * * * ?
+  collectBrokerStats:
+    cron: 15 0/1 * * * ?
+  collectBrokerRuntimeStats:
+    cron: 15 0/1 * * * ?
diff --git a/src/main/resources/logback.xml b/src/main/resources/logback.xml
index b5291d3..8032516 100644
--- a/src/main/resources/logback.xml
+++ b/src/main/resources/logback.xml
@@ -1,33 +1,32 @@
 <?xml version="1.0" encoding="UTF-8"?>
 <configuration>
-	<appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
-		<encoder charset="UTF-8">
-			<pattern>[%d{yyyy-MM-dd HH:mm:ss.SSS}] %5p %m%n</pattern>
-		</encoder>
-	</appender>
+    <appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
+        <encoder charset="UTF-8">
+            <pattern>[%d{yyyy-MM-dd HH:mm:ss.SSS}] %5p %m%n</pattern>
+        </encoder>
+    </appender>
 
-	<appender name="FILE"
-		class="ch.qos.logback.core.rolling.RollingFileAppender">
-		<file>${user.home}/logs/exporterlogs/rocketmq-exporter.log</file>
-		<append>true</append>
-		<rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
-			<fileNamePattern>${user.home}/logs/exporterlogs/rocketmq-exporter-%d{yyyy-MM-dd}.%i.log
-			</fileNamePattern>
-			<timeBasedFileNamingAndTriggeringPolicy
-				class="ch.qos.logback.core.rolling.SizeAndTimeBasedFNATP">
-				<maxFileSize>104857600</maxFileSize>
-			</timeBasedFileNamingAndTriggeringPolicy>
-			<MaxHistory>10</MaxHistory>
-		</rollingPolicy>
-		<encoder>
-			<pattern>[%d{yyyy-MM-dd HH:mm:ss.SSS}] %5p %m%n</pattern>
-			<charset class="java.nio.charset.Charset">UTF-8</charset>
-		</encoder>
-	</appender>
+    <appender name="FILE"
+              class="ch.qos.logback.core.rolling.RollingFileAppender">
+        <file>./logs/exporterlogs/rocketmq-exporter.log</file>
+        <append>true</append>
+        <rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
+            <fileNamePattern>./logs/exporterlogs/rocketmq-exporter-%d{yyyy-MM-dd}.%i.log
+            </fileNamePattern>
+            <timeBasedFileNamingAndTriggeringPolicy class="ch.qos.logback.core.rolling.SizeAndTimeBasedFNATP">
+                <maxFileSize>104857600</maxFileSize>
+            </timeBasedFileNamingAndTriggeringPolicy>
+            <totalSizeCap>20gb</totalSizeCap>
+        </rollingPolicy>
+        <encoder>
+            <pattern>[%d{yyyy-MM-dd HH:mm:ss.SSS}] %5p %m%n</pattern>
+            <charset class="java.nio.charset.Charset">UTF-8</charset>
+        </encoder>
+    </appender>
 
-	<root level="INFO">
-		<appender-ref ref="STDOUT" />
-		<appender-ref ref="FILE" />
-	</root>
+    <root level="INFO">
+        <appender-ref ref="STDOUT"/>
+        <appender-ref ref="FILE"/>
+    </root>
 
-</configuration> 
\ No newline at end of file
+</configuration>


Mime
View raw message