RocketMQ关于指标数据的总结

---

前言

在RocketMQ中有很多关于指标数据统计的实现，本篇文章就来总结下其实现的方式以及都统计了哪些数据（源码是4.9.1版本）。

---

一、BrokerStatsManager

1.BrokerStatsManager初始化及启动

（1）BrokerStatsManager初始化
在broker启动过程中会创建BrokerController对象，在初始化BrokerController时就会初始化BrokerStatsManager，具体如下：

public static BrokerController createBrokerController(String[] args) {
	...
	final BrokerController controller = new BrokerController(
        brokerConfig,
        nettyServerConfig,
        nettyClientConfig,
        messageStoreConfig);
            // remember all configs to prevent discard
    controller.getConfiguration().registerConfig(properties);
	...
}

//在构造BrokerController对象时会初始化BrokerStatsManager
public BrokerController(
        final BrokerConfig brokerConfig,
        final NettyServerConfig nettyServerConfig,
        final NettyClientConfig nettyClientConfig,
        final MessageStoreConfig messageStoreConfig
    ) {
    ...
	this.brokerStatsManager = new BrokerStatsManager(this.brokerConfig.getBrokerClusterName());
	...
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

在BrokerStatsManager中有一个map结构的statsTable是用于存储BrokerStatsManager统计的指标名称及其对应的指标数据，在BrokerStatsManager初始化过程中最重要的操作就是将其统计的所有指标放入这个map结构中。
（2）BrokerStatsManager启动
在启动BrokerController时会启动BrokerStatsManager，具体如下：

public static BrokerController start(BrokerController controller) {
        try {

            controller.start();

            String tip = "The broker[" + controller.getBrokerConfig().getBrokerName() + ", "
                + controller.getBrokerAddr() + "] boot success. serializeType=" + RemotingCommand.getSerializeTypeConfigInThisServer();

            if (null != controller.getBrokerConfig().getNamesrvAddr()) {
                tip += " and name server is " + controller.getBrokerConfig().getNamesrvAddr();
            }

            log.info(tip);
            System.out.printf("%s%n", tip);
            return controller;
        } catch (Throwable e) {
            e.printStackTrace();
            System.exit(-1);
        }

        return null;
    }

public void start() throws Exception {
	...
	if (this.brokerStatsManager != null) {
        this.brokerStatsManager.start();
    }
    ...
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

2.BrokerStatsManager运转

想要搞清楚BrokerStatsManager是如何统计数据的，需要搞清楚HashMap statsTable、StatsItemSet及StatsItem这三个数据结构。在统计指标数据时当然会有指标名称，在每个指标名称下还会有具体的对象，例如topic，最后才是该指标的值。所以在设计上首先是由statsTable来缓存所有的指标，每个指标对应的StatsItemSet中会有一个map结构statsItemTable用来缓存该指标下具体的对象及其指标值，指标值被封装成StatsItem。这三个数据结构详解具体如下。

2.1 statsTable

statsTable在BrokerStatsManager中是用于缓存统计的指标数据，statsTable的key表示统计的指标名称，value是StatsItemSet类型的数据，在BrokerStatsManager初始化时向statsTable中放入了统计的指标及其对应的StatsItemSet。

2.2 StatsItemSet

（1）StatsItemSet初始化
在BrokerStatsManager中每个指标都对应一条StatsItemSet类型的数据，其结构具体如下，其中statsName表示指标的名称，statsItemTable缓存的是该指标下具体的指标对象及其对应的StatsItem类型对象，举个例子，TOPIC_PUT_NUMS指标是在统计写入commitlog的消息数，此时会统计各个topic对应的消息数，所以topic的名称就是statsItemTable的key。

public class StatsItemSet {
    private final ConcurrentMap<String/* key */, StatsItem> statsItemTable =
        new ConcurrentHashMap<String, StatsItem>(128);

    private final String statsName;
    private final ScheduledExecutorService scheduledExecutorService;
    private final InternalLogger log;

    public StatsItemSet(String statsName, ScheduledExecutorService scheduledExecutorService, InternalLogger log) {
        this.statsName = statsName;
        this.scheduledExecutorService = scheduledExecutorService;
        this.log = log;
        this.init();
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

（2）init()
在StatsItemSet的初始化中有个很重要的方法就是init()，该方法中主要是一些定时任务，定时任务可以分为两大类，分别是samplingInxxx()和printAtxxx()。

• samplingInxxx()方法主要完成的功能是定时根据StatsItem中的指标值value、次数times以及当前的时间点构建一个指标的快照并将该指标快照存储在一个LinkedList中。
• printAtxxx()方法主要完成的功能是在broker端的日志文件stats.log中打印指标数据，按照定时任务的时间频率即每分钟、每小时、每天打印各指标数据，后面输出的指标数据SUM、TPS以及AVGPT是根据LinkedList中缓存的快照数据进行计算，具体计算逻辑见computeStatsData方法。

[statsName] [statsKey] Stats In One Minute, SUM: xxx TPS: xxx AVGPT: xxx
[statsName] [statsKey] Stats In One Hour, SUM: xxx TPS: xxx AVGPT: xxx
[statsName] [statsKey] Stats In One Day, SUM: xxx TPS: xxx AVGPT: xxx
1
2
3

private static StatsSnapshot computeStatsData(final LinkedList<CallSnapshot> csList) {
        StatsSnapshot statsSnapshot = new StatsSnapshot();
        synchronized (csList) {
            double tps = 0;
            double avgpt = 0;
            long sum = 0;
            long timesDiff = 0;
            if (!csList.isEmpty()) {
                CallSnapshot first = csList.getFirst();
                CallSnapshot last = csList.getLast();
                sum = last.getValue() - first.getValue();
                tps = (sum * 1000.0d) / (last.getTimestamp() - first.getTimestamp());

                timesDiff = last.getTimes() - first.getTimes();
                if (timesDiff > 0) {
                    avgpt = (sum * 1.0d) / timesDiff;
                }
            }

            statsSnapshot.setSum(sum);
            statsSnapshot.setTps(tps);
            statsSnapshot.setAvgpt(avgpt);
            statsSnapshot.setTimes(timesDiff);
        }

        return statsSnapshot;
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

2.3 StatsItem

StatsItem的数据结构如下，其中value表示指标值，times表示次数，csListMiute、csListHour和csListDay是用来缓存指标快照的，statsName表示指标名称，statsKey表示某一个指标下的具体对象。

public class StatsItem {

    private final AtomicLong value = new AtomicLong(0);

    private final AtomicLong times = new AtomicLong(0);

    private final LinkedList<CallSnapshot> csListMinute = new LinkedList<CallSnapshot>();

    private final LinkedList<CallSnapshot> csListHour = new LinkedList<CallSnapshot>();

    private final LinkedList<CallSnapshot> csListDay = new LinkedList<CallSnapshot>();

    private final String statsName;
    private final String statsKey;
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

3.统计的指标及其含义

二、StoreStatsService

StoreStatsService继承了ServiceThread，其统计的是和存储相关的指标数据。由于它本质上是个线程，所以我们直接看其run方法，可以看到它每1分钟会执行sampling和printTps方法。其中sampling方法主要完成的功能是往putTimesList、getTimesFoundList、getTimesMissList和transferedMsgCountList中添加指标的快照数据。

public void run() {
        log.info(this.getServiceName() + " service started");

        while (!this.isStopped()) {
            try {
                this.waitForRunning(FREQUENCY_OF_SAMPLING);

                this.sampling();

                this.printTps();
            } catch (Exception e) {
                log.warn(this.getServiceName() + " service has exception. ", e);
            }
        }

        log.info(this.getServiceName() + " service end");
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

sampling方法中存储各指标快照的含义具体如下：

printTps方法会每1分钟在日志中打印以下信息：

[STORETPS] put_tps XXX get_found_tps XXX get_miss_tps XXX get_transfered_tps XXX

StoreStatsService - [PAGECACHERT] TotalPut 0, PutMessageDistributeTime [<=0ms]:xxx [0~10ms]:xxx [10~50ms]:xxx [50~100ms]:xxx [100~200ms]:xxx [200~500ms]:xxx [500ms~1s]:xxx [1~2s]:xxx [2~3s]:xxx [3~4s]:xxx [4~5s]:xxx [5~10s]:xxx [10s~]:xxx
1
2
3

---

三、BrokerStats

在创建BrokerController时会初始化BrokerStats，并且还会周期性执行record方法，周期是一天一次，具体如下：

public boolean initialize() throws CloneNotSupportedException {
        boolean result = this.topicConfigManager.load();

        result = result && this.consumerOffsetManager.load();
        result = result && this.subscriptionGroupManager.load();
        result = result && this.consumerFilterManager.load();

        if (result) {
            try {
                this.messageStore =
                    new DefaultMessageStore(this.messageStoreConfig, this.brokerStatsManager, this.messageArrivingListener,
                        this.brokerConfig);
                if (messageStoreConfig.isEnableDLegerCommitLog()) {
                    DLedgerRoleChangeHandler roleChangeHandler = new DLedgerRoleChangeHandler(this, (DefaultMessageStore) messageStore);
                    ((DLedgerCommitLog)((DefaultMessageStore) messageStore).getCommitLog()).getdLedgerServer().getdLedgerLeaderElector().addRoleChangeHandler(roleChangeHandler);
                }
                this.brokerStats = new BrokerStats((DefaultMessageStore) this.messageStore);
                ...
            } catch (IOException e) {
                result = false;
                log.error("Failed to initialize", e);
            }
        }
        ...

        if (result) {
            ...
            final long initialDelay = UtilAll.computeNextMorningTimeMillis() - System.currentTimeMillis();
            final long period = 1000 * 60 * 60 * 24;
            this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
                @Override
                public void run() {
                    try {
                        BrokerController.this.getBrokerStats().record();
                    } catch (Throwable e) {
                        log.error("schedule record error.", e);
                    }
                }
            }, initialDelay, period, TimeUnit.MILLISECONDS);

			...
        }
        return result;
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44

record方法主要是在日志中输出以下信息：

yesterday put message total: xxx （其含义是昨天成功写入broker中的消息总条数）
yesterday get message total: xxx （其含义是昨天成功broker中获取消息的次数）
1
2

BrokerStats中包含的指标信息如下：

public void record() {
        this.msgPutTotalYesterdayMorning = this.msgPutTotalTodayMorning;
        this.msgGetTotalYesterdayMorning = this.msgGetTotalTodayMorning;

        this.msgPutTotalTodayMorning =
            this.defaultMessageStore.getStoreStatsService().getPutMessageTimesTotal();
        this.msgGetTotalTodayMorning =
            this.defaultMessageStore.getStoreStatsService().getGetMessageTransferedMsgCount().get();

        log.info("yesterday put message total: {}", msgPutTotalTodayMorning - msgPutTotalYesterdayMorning);
        log.info("yesterday get message total: {}", msgGetTotalTodayMorning - msgGetTotalYesterdayMorning);
    }
1
2
3
4
5
6
7
8
9
10
11
12

四、RocketMQ-Exporter

在RocketMQ-Exporter中有5个定时任务来获取指标数据，定时任务的周期是每分钟一次并且是在每分钟的第15秒开始执行，定时任务的周期可以在其配置文件中进行修改。

task:
  count: 5 # num of scheduled-tasks
  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 * * * ?
1
2
3
4
5
6
7
8
9
10
11
12

RocketMQ-Exporter中的指标及其含义具体如下：