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.总览:
这是一张非常重要的图
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 原理和源码分析的主要内容,如果未能解决你的问题,请参考以下文章