本文阅读基础：使用或了解过rocketMq；想了解"topic 5min trend"背后的原理；想了解监控模式如何实现。

RocketMq的dashboard，有运维页面，驾驶舱，集群页面，主题页面，消费者页面，生产者页面，发布管理页面，消息查询页面等，为开发和运维提供了强大的监控功能。

中文文档：https://github.com/apache/rocketmq-dashboard/blob/master/docs/1_0_0/UserGuide_CN.md

这篇文章将介绍驾驶舱-topic 5min trend的实现和原理。

功能：

"topic 5min trend"表示当前时间/或选定时间，25小时内，rocketMq消息发送总量，如果是当天实时消息，可能有1h延迟。

原理：

1.整体分为3个模块：

1、broker起定时任务收集消息-核心；

2、dashboard起定时任务从broker拉取统计内容，汇总并保存到本地；

3、前端拉取消息并在本地创建当天统计文件；

2.代码版本：

broker：4.2.0

dashboard：1.0.1-SNAPSHOT

3.总览：

这是一张非常重要的图

dashboard topic 5min trend活动图

4.源码分析：

我们从右往左看这张图，右边是前端调用，左边是数据收集。

整体上，dashboard通过定时任务，从broker拉取统计数据，按天保存到本地。

第6、7泳道，是将数据保存到本地，提供给前端页面展示；

第5条泳道中，生成sum，我们主要看下sum的生成逻辑：先取statsDay，如果没有，再依次取statsHour和statsMinute；

看到这里有点懵，brokerStatsData是从哪来的？又是如何处理的呢？我们看第3，4条泳道，也就是本文的重点。

3号泳道，broker的StatsItemSet中，通过执行定时任务对写入的消息量采样汇总到StatsItem。

4号泳道，broker的AdminBrokerProcessor分别对StatsItem中的3个成员变量csListMinute,csListHour,csListDay处理，生成3个快照汇总到BrokerStatsData。

下面我们进入源码，看下3号泳道和4号泳道的实现原理。

3号泳道，broker-StatsItemSet 采样汇总

StatsItemSet类在实例化时，执行init方法，启动定时任务

具体的采样方法：

从图中我们看出，csListMinute,csListHour,csListDay，用来存放采样的数据。

csListMinute存放最近70s的数据快照；csListHour存放最近70min的数据快照；csListDay存放最近25h的数据快照。

同时，我们看到，三个取样器取的总量是来自同一个value：

value是写入的消息总量；times是写入次数；驾驶舱-topic 5min trend中的曲线表示的是消息总量，所以我们主要看value。

看到这里，大家可能和我有同样的疑惑，都取同一个值，要这么多采样器做什么？我们去找调用方看看怎么处理这些数据。

4号泳道：broker-AdminBrokerProcessor类ViewBrokerStatsData方法：

private RemotingCommand ViewBrokerStatsData(ChannelHandlerContext ctx,
        RemotingCommand request) throws RemotingCommandException {
        final ViewBrokerStatsDataRequestHeader requestHeader =
            (ViewBrokerStatsDataRequestHeader) request.decodeCommandCustomHeader(ViewBrokerStatsDataRequestHeader.class);
        final RemotingCommand response = RemotingCommand.createResponseCommand(null);
        DefaultMessageStore messageStore = (DefaultMessageStore) this.brokerController.getMessageStore();
        // 从Map中，根据topic获取statsItem
        StatsItem statsItem = messageStore.getBrokerStatsManager().getStatsItem(requestHeader.getStatsName(), requestHeader.getStatsKey());
        if (null == statsItem) {
            response.setCode(ResponseCode.SYSTEM_ERROR);
            response.setRemark(String.format("The stats <%s> <%s> not exist", requestHeader.getStatsName(), requestHeader.getStatsKey()));
            return response;
        }

        BrokerStatsData brokerStatsData = new BrokerStatsData();

        {
            BrokerStatsItem it = new BrokerStatsItem();
            // StatsDataInMinute 有计算方法
            StatsSnapshot ss = statsItem.getStatsDataInMinute();
            it.setSum(ss.getSum());
            it.setTps(ss.getTps());
            it.setAvgpt(ss.getAvgpt());
            brokerStatsData.setStatsMinute(it);
        }

        {
            BrokerStatsItem it = new BrokerStatsItem();
            StatsSnapshot ss = statsItem.getStatsDataInHour();
            it.setSum(ss.getSum());
            it.setTps(ss.getTps());
            it.setAvgpt(ss.getAvgpt());
            brokerStatsData.setStatsHour(it);
        }

        {
            BrokerStatsItem it = new BrokerStatsItem();
            StatsSnapshot ss = statsItem.getStatsDataInDay();
            it.setSum(ss.getSum());
            it.setTps(ss.getTps());
            it.setAvgpt(ss.getAvgpt());
            brokerStatsData.setStatsDay(it);
        }

        response.setBody(brokerStatsData.encode());
        response.setCode(ResponseCode.SUCCESS);
        response.setRemark(null);
        return response;
    }

dashboard调用broker的ViewBrokerStatsData()方法，获取统计快照BrokerStatsData。

BrokerStatsData类继承了RemotingSerializable类，有3个成员变量：statsMinute,statsHour,statsDay。

我们看下从哪里对这三个变量赋值的，以statsMinute为例：

快照ss来自于statsItem.getStatsDataInMinute()：

从2,3两图我们看到，计算statsMinute快照时，computeStatsData的入参是csListMinute，就是我们采样的结果集。

computeStatsData是静态公共的计算方法，分别对csListMinute,csListHour,csListDay处理，计算出分钟级数据快照statsDataInMinute，小时级数据快照statsDataInHour，天级数据快照statsDataInDay；再将这些快照中的值，赋值给BrokerStatsItem和BrokerStatsData。

到这里，我们就梳理清楚了BrokerStatsData中的数据是怎么来的了，以及数据是如何处理的了。

下面我本地debug，结合dashboard来展开说下图3中的数据处理规则。已经明白的朋友可以跳过

5.图表和数据：

我们看到对csList写入和获取时，都对csList加锁。用于防止多线程请求下，csList动态变化产生数据不一致问题。因为用于监控，最快10s写入一次，调用方30s获取一次，所以可以忽略加锁对性能产生的影响。

计算sum时，是用最后一个节点的value-第一个节点的value得到。

这是我本地发送消息后的一段截图，我们debug看下，3000是怎么来的？

在文章开始处，代码分析，第5条泳道中，我们说过，sum的生成逻辑：从brokerStatsData中先取statsDay，如果没有，再依次取statsHour和statsMinute；所以，图中的3000，取自statsDay中的sum。

按照我们上面的推理，statsDay中sum，是用csListDay中的最后一个节点的value-第一个节点的value得到。

我们进入statsItem，看下是不是这样：

6000-3000=3000；

Q&A：

1、Q：为什么csListDay第一个节点到最后一个节点隔了10个小时左右，却只有5个节点。

A：我猜是我打断点或者电脑休眠导致的。你看，4和3节点之间间隔是1个小时的；按照代码的逻辑，一直运行的话，应该是25个节点。

2、Q：几天前统计的数据，是怎么保存的？

A：在文章开头的活动图中，泳道1中可以看到，收集到的数据会实时保存到本地文件。dashboard请求时，也是请求本地文件。将本地文件中的数据转成Map返回。

public Map<String, List<String>> getTopicCache(String date) {
        // 获取文件目录：/tmp/rocketmq-console/data/dashboard
        String dataLocationPath = configure.getDashboardCollectData();
        // 生成文件：/tmp/rocketmq-console/data/dashboardyyyy-MM-dd_topic.json
        File file = new File(dataLocationPath + date + "_topic" + ".json");
        if (!file.exists()) {
            log.info(String.format("No dashboard data for data: %s", date));
            //throw Throwables.propagate(new ServiceException(1, "This date have't data!"));
            return Maps.newHashMap();
        }
        return jsonDataFile2map(file);
    }

/tmp/rocketmq-console/data目录：

3、Q：标题叫驾驶舱-topic 5min trend。标题和分析的结果出入很大呢，5min没体现啊。

A：我分析的是4.2版本的代码，逻辑确实就是上面分析的结果，而且我结合dashboard反复验证过。

比如上面我们说，sum的生成逻辑：从brokerStatsData中先取statsDay，如果没有，再依次取statsHour和statsMinute；

也就是说，当statsDay不为空时，1个小时内即使有消息产生，只要没到samplingInHour任务执行的时间，在驾驶舱-topic 5min trend中就不会体现出来。

黄线标识的时刻我没有发送消息，从“Broker 5min trend”中可以看出，但“topic 5min trend”却上升了，是因为我本地的samplingInHour执行了，拿到了我在15:02到16:02的sum。在16:05和16:07左右，我分别发送了1000条消息，“主题 5min trend”却没有变化，预计在17:02时，才会上升。

4、Q：我们知道rocketMq可以集群部署，多主多从，一主多从。我们看到消息采集是分布在每个broker主节点上。那dashboard是如何收集到这些消息的呢？

A：非常简单，dashboard从namesrv可以获得到所有注册的broker主节点。从这些主节点挨个获取再加和就可以了。