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

简介: 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延迟。



e361b8818e7e0bc3b615b6fc722ca6fe.png


原理:

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;



d11dc10a1726127885376d58c06227d3.png


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


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


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


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


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



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

image.png


具体的采样方法:



6f679304033110524ed681ac98206ddb.png


image.png


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


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


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




image.png


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。



eef13b9d7798101f143a20fd02d10685.png


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


image.png


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


image.png


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


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


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


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



5.图表和数据:




image.png


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


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


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



image.png


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



image.png




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

我们进入statsItem,看下是不是这样:



image.png



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目录:


image.png


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


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


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


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


image.png


image.png


黄线标识的时刻我没有发送消息,从“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主节点。从这些主节点挨个获取再加和就可以了。



image.png


总结:


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

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

5min 没有体现出来。




















相关实践学习
RocketMQ一站式入门使用
从源码编译、部署broker、部署namesrv,使用java客户端首发消息等一站式入门RocketMQ。
消息队列 MNS 入门课程
1、消息队列MNS简介 本节课介绍消息队列的MNS的基础概念 2、消息队列MNS特性 本节课介绍消息队列的MNS的主要特性 3、MNS的最佳实践及场景应用 本节课介绍消息队列的MNS的最佳实践及场景应用案例 4、手把手系列:消息队列MNS实操讲 本节课介绍消息队列的MNS的实际操作演示 5、动手实验:基于MNS,0基础轻松构建 Web Client 本节课带您一起基于MNS,0基础轻松构建 Web Client
相关文章
|
5月前
|
消息中间件 RocketMQ
RocketMQ报错:MQClientException:no route info of this topic的解决
RocketMQ报错:MQClientException:no route info of this topic的解决
120 0
|
30天前
|
消息中间件 物联网 网络性能优化
MQTT常见问题之MQTT的topic超出上限25个如何解决
MQTT(Message Queuing Telemetry Transport)是一个轻量级的、基于发布/订阅模式的消息协议,广泛用于物联网(IoT)中设备间的通信。以下是MQTT使用过程中可能遇到的一些常见问题及其答案的汇总:
|
3月前
|
消息中间件 网络架构
【面试问题】什么是 MQ topic 交换器(模式匹配) ?
【1月更文挑战第27天】【面试问题】什么是 MQ topic 交换器(模式匹配) ?
|
4月前
|
消息中间件 Java Kafka
RabbitMQ安装和5种不同的消息模型(BasicQueue,WorkQueue,Fanout Exchange,Direct Exchange,Topic Exchange)与SpringAMQP
RabbitMQ安装和5种不同的消息模型(BasicQueue,WorkQueue,Fanout Exchange,Direct Exchange,Topic Exchange)与SpringAMQP
|
8月前
|
消息中间件 Java RocketMQ
Rocketmq创建topic报错org.apache.rocketmq.acl.common.AclException: [10015:signature-failed] unable to cal
Rocketmq创建topic报错org.apache.rocketmq.acl.common.AclException: [10015:signature-failed] unable to cal
331 0
|
9月前
|
消息中间件 应用服务中间件 nginx
【RabbitMQ六】——RabbitMQ主题模式(Topic)
【RabbitMQ六】——RabbitMQ主题模式(Topic)
165 1
|
11月前
|
消息中间件
七、RabbitMQ 之 SpringAMQP 实现 Topic 交换机
七、RabbitMQ 之 SpringAMQP 实现 Topic 交换机
|
11月前
|
消息中间件 JSON 运维
Rocket MQ报错No route info of this topic的问题探究
Rocket MQ报错No route info of this topic的问题探究
616 0
|
消息中间件 存储 JSON
【RabbitMQ】Fanout、Direct、Topic、消息转换器
【RabbitMQ】Fanout、Direct、Topic、消息转换器
213 0
【RabbitMQ】Fanout、Direct、Topic、消息转换器

热门文章

最新文章