RocketMq-dashboard:topic 5min trend 原理和源码分析

Posted 小王师傅66

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了RocketMq-dashboard:topic 5min trend 原理和源码分析相关的知识,希望对你有一定的参考价值。

本文阅读基础:使用或了解过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主节点。从这些主节点挨个获取再加和就可以了。

总结:

到此,我们就将"topic 5min trend"中主要的实现结合源码,图表展示和数据分析完了。

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

5min 没有体现出来。

以上是关于RocketMq-dashboard:topic 5min trend 原理和源码分析的主要内容,如果未能解决你的问题,请参考以下文章

docker安装rocketmq 开通acl鉴权 rocketmq-dashboard

kafka删除topic

如何在kafka中创建topic

kafka 常见命令以及增加topic的分区数

kafka 彻底删除topic

95-35-010-Topic-Topic 扩容 源码解析