RocketMq之一条消息在commitlog文件中如何存储验证

本文的目的在于记录本次学习过程，在看《RocketMQ技术内幕》一书，关于消息存储，时，看到关于计算消息总长度的方法，着迷了，想要对CommitLog文件中存储的信息进行分析。

一条消息存储到commitlog文件中的总长度计算方式（源码）：

// 包路径
org.apache.rocketmq.store.CommitLog#calMsgLength// 计算消息长度// CommitLog条目是不定长的，每一个条目的长度存储在前4个字节中protected static int calMsgLength(int sysFlag, int bodyLength, int topicLength, int propertiesLength) {int bornhostLength = (sysFlag & MessageSysFlag.BORNHOST_V6_FLAG) == 0 ? 8 : 20;int storehostAddressLength = (sysFlag & MessageSysFlag.STOREHOSTADDRESS_V6_FLAG) == 0 ? 8 : 20;final int msgLen = 4 //TOTALSIZE，该消息总长度4字节+ 4 //MAGICCODE，魔数，4字节，固定值0xdaa320a7+ 4 //BODYCRC，消息体CRC校验码，4字节+ 4 //QUEUEID，消息消费队列ID，4字节+ 4 //FLAG，消息Flag，ROcketMQ不做处理，供用户自定义使用默认4字节。+ 8 //QUEUEOFFSET，消息在消息消费队列的偏移量8字节。+ 8 //PHYSICALOFFSET，消息在消息消费队列的偏移量+ 4 //SYSFLAG，消息系统Flag，例如是否压缩、是否是事务消息等，4字节。+ 8 //BORNTIMESTAMP，生产者调用消息发送API的时间戳，8字节。+ bornhostLength //BORNHOST，消息发送者IP、端口号，8字节+ 8 //STORETIMESTAMP，消息存储时间戳8字节。+ storehostAddressLength //STOREHOSTADDRESS，Broker服务器IP+端口号，8字节+ 4 //RECONSUMETIMES，消息重试次数，4字节。+ 8 //Prepared Transaction Offset事务消息物理偏移量8字节+ 4 + (bodyLength > 0 ? bodyLength : 0) //BODY，消息体长度（定义4字节）+ 消息长度+ 1 + topicLength //TOPIC，主题存储长度1字节，主题内容长度+ 2 + (propertiesLength > 0 ? propertiesLength : 0) //propertiesLength，消息属性长度，表示消息属性长度不能超过65536个字符+ 0;return msgLen;}

结合上面的源码，我读取一个commitlog文件，进行分析。由于commitlog只存储消息，因此该文件从第一个字节开始就存储了消息内容。待读取的commitlog文件位置：

分析commitlog文件的代码：

package com.an.store;import cn.hutool.core.date.DateUnit;
import cn.hutool.core.date.DateUtil;
import cn.hutool.core.util.HexUtil;import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.nio.file.StandardOpenOption;
import java.util.Arrays;// 存储文件解密
public class CommitLogTest {public static void main(String[] args) throws IOException {String fileName_or_phyficalOffset = "00000000001073741824";Path path = Paths.get("E:\\tmp\\rocketmq-data-log\\broker-a-0\\data\\store\\commitlog", fileName_or_phyficalOffset);FileChannel fileChannel = FileChannel.open(path, StandardOpenOption.READ);// 设大一点，尽量一次读取完一条消息的完整字节。ByteBuffer byteBuffer = ByteBuffer.allocate(2048);fileChannel.read(byteBuffer);// 输出commitLog的信息byteBuffer.flip();// 记录消息总长度，每一次操作记录一次，最终与读取的消息总长度对比int totalLength = 0;int totalLengthF = byteBuffer.getInt();println("TOTALSIZE，该消息总长度4字节",totalLengthF);totalLength+=4;byte[] magic = new byte[4];byteBuffer.get(magic);println("MAGICCODE，魔数，4字节，固定值0xdaa320a7",HexUtil.encodeHexStr(magic));totalLength+=4;readByte2Str(byteBuffer,"BODYCRC，消息体CRC校验码，4字节",4);totalLength+=4;println("QUEUEID，消息消费队列ID，4字节",byteBuffer.getInt());totalLength+=4;readByte2Str(byteBuffer,"//FLAG，消息Flag，ROcketMQ不做处理，供用户自定义使用默认4字节。",4);totalLength+=4;println("QUEUEOFFSET，消息在消息消费队列的偏移量8字节",byteBuffer.getLong());totalLength+=8;println("PHYSICALOFFSET，消息在消息消费队列的偏移量",byteBuffer.getLong());totalLength+=8;System.out.println("文件名:" + fileName_or_phyficalOffset);SYSFLAG，消息系统Flag，例如是否压缩、是否是事务消息等，4字节// 不知具体的原始类型。。。。readByte2Str(byteBuffer,"//SYSFLAG，消息系统Flag，例如是否压缩、是否是事务消息等，4字节",4);totalLength+=4;//  8 //BORNTIMESTAMP，生产者调用消息发送API的时间戳，8字节。long bornTimestamp = byteBuffer.getLong();String timePattern = "yyyy-MM-dd HH:mm:ss.sss";println("BORNTIMESTAMP，生产者调用消息发送API的时间戳，8字节。",DateUtil.date(bornTimestamp).toString(timePattern));totalLength+=8;// bornhostLength //BORNHOST，消息发送者IP、端口号，8字节byte[] ipbyte4s = new byte[4];byteBuffer.get(ipbyte4s);println("bornhostLength //BORNHOST，消息发送者IP、端口号，8字节---------->IP",Arrays.toString(ipbyte4s));println("bornhostLength //BORNHOST，消息发送者IP、端口号，8字节---------->PORT",byteBuffer.getInt());totalLength+=8;// //STORETIMESTAMP，消息存储时间戳8字节。long storeTimestamp = byteBuffer.getLong();println("//STORETIMESTAMP，消息存储时间戳8字节。",DateUtil.date(storeTimestamp).toString(timePattern));totalLength+=8;// storehostAddressLength //STOREHOSTADDRESS，Broker服务器IP+端口号，8字节byte[] ipByte = new byte[4];byteBuffer.get(ipByte);println("storehostAddressLength //STOREHOSTADDRESS，Broker服务器IP+端口号，8字节.---->IP", Arrays.toString(ipByte));int port = byteBuffer.getInt();println("storehostAddressLength //STOREHOSTADDRESS，Broker服务器IP+端口号，8字节.---->PORT",port);totalLength+=8;//  //RECONSUMETIMES，消息重试次数，4字节。int retryTimes = byteBuffer.getInt();println("RECONSUMETIMES，消息重试次数，4字节",retryTimes);totalLength+=4;// Prepared Transaction Offset事务消息物理偏移量8字节println("Prepared Transaction Offset事务消息物理偏移量8字节",byteBuffer.getLong());totalLength+=8;// 4 + (bodyLength > 0 ? bodyLength : 0) //BODY，消息体长度（定义4字节）+ 消息长度int bodyLength = byteBuffer.getInt();println("4 + (bodyLength > 0 ? bodyLength : 0) //BODY，消息体长度（定义4字节）+ 消息长度.----->内容长度",bodyLength);readByte2StrNormalPrint(byteBuffer,"4 + (bodyLength > 0 ? bodyLength : 0) //BODY，消息体长度（定义4字节）+ 消息长度.----->内容",bodyLength);totalLength+=4+bodyLength;// 1 + topicLength //TOPIC，主题存储长度1字节，主题内容长度System.out.println("----------------------");byte topicLen = byteBuffer.get();System.out.println("1 + topicLength //TOPIC，主题存储长度1字节，主题内容长度------->topic的长度：" + (int)topicLen);readByte2StrNormalPrint(byteBuffer,"1 + topicLength //TOPIC，主题存储长度1字节，主题内容长度---->topic",(int)topicLen);totalLength+=1+(int)topicLen;//2 + (propertiesLength > 0 ? propertiesLength : 0) //propertiesLength，消息属性长度，表示消息属性长度不能超过65536个字符short propertyLen = byteBuffer.getShort();System.out.println("----------------------");System.out.println("2 + (propertiesLength > 0 ? propertiesLength : 0) //propertiesLength，消息属性长度，表示消息属性长度不能超过65536个字符,长度：" + propertyLen);readByte2StrNormalPrint(byteBuffer,"2 + (propertiesLength > 0 ? propertiesLength : 0) //propertiesLength，消息属性长度，表示消息属性长度不能超过65536个字符,内容",(int)propertyLen);totalLength+=2+propertyLen;System.out.println("------------------------------一条消息探索完毕------------------------------------");System.out.println("消息长度（文件,totalLengthF）："+ totalLengthF );System.out.println("消息长度（计算,totalLength）：" + totalLength);fileChannel.close();}public static void readByte2Str(ByteBuffer byteBuffer,String key,int len){byte[] value = new byte[len];byteBuffer.get(value);println(key,new String(value));}public static void readByte2StrNormalPrint(ByteBuffer byteBuffer,String key,int len){byte[] value = new byte[len];byteBuffer.get(value);printlnNormal(key,new String(value));}public static void println(String key,Object value){System.out.println("--------------------");System.out.println("Key["+key+"],value["+value+"]");}public static void printlnNormal(String key,Object value){System.out.println("Key["+key+"],value["+value+"]");}}

运行结果分析

详细运行结果：

E:\soft\Java\jdk1.8.0_171\bin\java.exe "-javaagent:E:\soft\JetBrains\IntelliJ IDEA 2019.3.3\lib\idea_rt.jar=62037:E:\soft\JetBrains\IntelliJ IDEA 2019.3.3\bin" -Dfile.encoding=UTF-8 -classpath E:\soft\Java\jdk1.8.0_171\jre\lib\charsets.jar;E:\soft\Java\jdk1.8.0_171\jre\lib\deploy.jar;E:\soft\Java\jdk1.8.0_171\jre\lib\ext\access-bridge-64.jar;E:\soft\Java\jdk1.8.0_171\jre\lib\ext\cldrdata.jar;E:\soft\Java\jdk1.8.0_171\jre\lib\ext\dnsns.jar;E:\soft\Java\jdk1.8.0_171\jre\lib\ext\jaccess.jar;E:\soft\Java\jdk1.8.0_171\jre\lib\ext\jfxrt.jar;E:\soft\Java\jdk1.8.0_171\jre\lib\ext\localedata.jar;E:\soft\Java\jdk1.8.0_171\jre\lib\ext\nashorn.jar;E:\soft\Java\jdk1.8.0_171\jre\lib\ext\sunec.jar;E:\soft\Java\jdk1.8.0_171\jre\lib\ext\sunjce_provider.jar;E:\soft\Java\jdk1.8.0_171\jre\lib\ext\sunmscapi.jar;E:\soft\Java\jdk1.8.0_171\jre\lib\ext\sunpkcs11.jar;E:\soft\Java\jdk1.8.0_171\jre\lib\ext\zipfs.jar;E:\soft\Java\jdk1.8.0_171\jre\lib\javaws.jar;E:\soft\Java\jdk1.8.0_171\jre\lib\jce.jar;E:\soft\Java\jdk1.8.0_171\jre\lib\jfr.jar;E:\soft\Java\jdk1.8.0_171\jre\lib\jfxswt.jar;E:\soft\Java\jdk1.8.0_171\jre\lib\jsse.jar;E:\soft\Java\jdk1.8.0_171\jre\lib\management-agent.jar;E:\soft\Java\jdk1.8.0_171\jre\lib\plugin.jar;E:\soft\Java\jdk1.8.0_171\jre\lib\resources.jar;E:\soft\Java\jdk1.8.0_171\jre\lib\rt.jar;E:\work\java15\my_rocketmq\example\target\classes;E:\work\java15\my_rocketmq\client\target\classes;E:\work\java15\my_rocketmq\common\target\classes;E:\mavenlib3.5-2\org\apache\commons\commons-lang3\3.4\commons-lang3-3.4.jar;E:\work\java15\my_rocketmq\srvutil\target\classes;E:\work\java15\my_rocketmq\remoting\target\classes;E:\mavenlib3.5-2\com\alibaba\fastjson\1.2.61\fastjson-1.2.61.jar;E:\mavenlib3.5-2\io\netty\netty-all\4.0.42.Final\netty-all-4.0.42.Final.jar;E:\mavenlib3.5-2\io\netty\netty-tcnative-boringssl-static\1.1.33.Fork26\netty-tcnative-boringssl-static-1.1.33.Fork26.jar;E:\mavenlib3.5-2\commons-cli\commons-cli\1.2\commons-cli-1.2.jar;E:\mavenlib3.5-2\ch\qos\logback\logback-classic\1.0.13\logback-classic-1.0.13.jar;E:\mavenlib3.5-2\ch\qos\logback\logback-core\1.0.13\logback-core-1.0.13.jar;E:\mavenlib3.5-2\org\slf4j\slf4j-api\1.7.7\slf4j-api-1.7.7.jar;E:\mavenlib3.5-2\org\javassist\javassist\3.20.0-GA\javassist-3.20.0-GA.jar;E:\mavenlib3.5-2\io\openmessaging\openmessaging-api\0.3.1-alpha\openmessaging-api-0.3.1-alpha.jar;E:\work\java15\my_rocketmq\openmessaging\target\classes;E:\work\java15\my_rocketmq\acl\target\classes;E:\work\java15\my_rocketmq\logging\target\classes;E:\mavenlib3.5-2\org\yaml\snakeyaml\1.19\snakeyaml-1.19.jar;E:\mavenlib3.5-2\commons-codec\commons-codec\1.9\commons-codec-1.9.jar;E:\mavenlib3.5-2\commons-validator\commons-validator\1.6\commons-validator-1.6.jar;E:\mavenlib3.5-2\commons-beanutils\commons-beanutils\1.9.2\commons-beanutils-1.9.2.jar;E:\mavenlib3.5-2\commons-digester\commons-digester\1.8.1\commons-digester-1.8.1.jar;E:\mavenlib3.5-2\commons-logging\commons-logging\1.2\commons-logging-1.2.jar;E:\mavenlib3.5-2\commons-collections\commons-collections\3.2.2\commons-collections-3.2.2.jar;E:\mavenlib3.5-2\cn\hutool\hutool-all\5.0.5\hutool-all-5.0.5.jar com.an.store.CommitLogTest
--------------------
Key[TOTALSIZE，该消息总长度4字节],value[1166]
--------------------
Key[MAGICCODE，魔数，4字节，固定值0xdaa320a7],value[daa320a7]
--------------------
Key[BODYCRC，消息体CRC校验码，4字节],value[sj6]
--------------------
Key[QUEUEID，消息消费队列ID，4字节],value[17]
--------------------
Key[//FLAG，消息Flag，ROcketMQ不做处理，供用户自定义使用默认4字节。],value[    ]
--------------------
Key[QUEUEOFFSET，消息在消息消费队列的偏移量8字节],value[920934]
--------------------
Key[PHYSICALOFFSET，消息在消息消费队列的偏移量],value[1073741824]
文件名:00000000001073741824
--------------------
Key[//SYSFLAG，消息系统Flag，例如是否压缩、是否是事务消息等，4字节],value[    ]
--------------------
Key[BORNTIMESTAMP，生产者调用消息发送API的时间戳，8字节。],value[2020-03-27 12:30:46.046]
--------------------
Key[bornhostLength //BORNHOST，消息发送者IP、端口号，8字节---------->IP],value[[127, 0, 0, 1]]
--------------------
Key[bornhostLength //BORNHOST，消息发送者IP、端口号，8字节---------->PORT],value[54256]
--------------------
Key[//STORETIMESTAMP，消息存储时间戳8字节。],value[2020-03-27 12:30:47.047]
--------------------
Key[storehostAddressLength //STOREHOSTADDRESS，Broker服务器IP+端口号，8字节.---->IP],value[[127, 0, 0, 1]]
--------------------
Key[storehostAddressLength //STOREHOSTADDRESS，Broker服务器IP+端口号，8字节.---->PORT],value[10911]
--------------------
Key[RECONSUMETIMES，消息重试次数，4字节],value[0]
--------------------
Key[Prepared Transaction Offset事务消息物理偏移量8字节],value[0]
--------------------
Key[4 + (bodyLength > 0 ? bodyLength : 0) //BODY，消息体长度（定义4字节）+ 消息长度.----->内容长度],value[898]
Key[4 + (bodyLength > 0 ? bodyLength : 0) //BODY，消息体长度（定义4字节）+ 消息长度.----->内容],value[春江潮水连海平，海上明月共潮生。滟滟随波千万里，何处春江无月明！江流宛转绕芳甸，月照花林皆似霰；空里流霜不觉飞，汀上白沙看不见。江天一色无纤尘，皎皎空中孤月轮。江畔何人初见月？江月何年初照人？人生代代无穷已，江月年年望相似。不知江月待何人，但见长江送流水。白云一片去悠悠，青枫浦上不胜愁。谁家今夜扁舟子？何处相思明月楼？可怜楼上月徘徊，应照离人妆镜台。玉户帘中卷不去，捣衣砧上拂还来。此时相望不相闻，愿逐月华流照君。鸿雁长飞光不度，鱼龙潜跃水成文。昨夜闲潭梦落花，可怜春半不还家。江水流春去欲尽，江潭落月复西斜。斜月沉沉藏海雾，碣石潇湘无限路。不知乘月几人归，落月摇情满江树。]
----------------------
1 + topicLength //TOPIC，主题存储长度1字节，主题内容长度------->topic的长度：19
Key[1 + topicLength //TOPIC，主题存储长度1字节，主题内容长度---->topic],value[SCHEDULE_TOPIC_XXXX]
----------------------
2 + (propertiesLength > 0 ? propertiesLength : 0) //propertiesLength，消息属性长度，表示消息属性长度不能超过65536个字符,长度：158
Key[2 + (propertiesLength > 0 ? propertiesLength : 0) //propertiesLength，消息属性长度，表示消息属性长度不能超过65536个字符,内容],value[REAL_TOPICTopicTestKEYS春江花月夜 [919935] ---> 2020:03:27 12:30:46UNIQ_KEYA9FE804E000018B4AAC28894B4D7097FWAITtrueDELAY18TAGSTagAREAL_QID0]
------------------------------一条消息探索完毕------------------------------------
消息长度（文件,totalLengthF）：1166
消息长度（计算,totalLength）：1166Process finished with exit code 0

总结
本次主要根据《RocketMQ技术内幕》这本书为依据，对存储在commitLOg文件中的内容进行了探索。可以从控制台打印中，明显感觉到与RocketMQ设计的吻合。比如文件名字与解析后得到的消息物理存储偏移量一致。本文crc这种没有正确的解析，这种可以使用RocketMQ提供的工具进行解析。还有用到了糊涂的工具包，用来解析2进制到16进制。

RocketMq之一条消息在commitlog文件中如何存储验证相关推荐

10-8 猫和狗：创建两个文件cats.txt和dogs.txt，在第一个文件中至少存储三只猫的名字，在第二个文件中至少存储三条狗的名字。编写一个程序，尝试读取这些文件，并将其内容打印到
10-8 猫和狗 : 创建两个文件cats.txt和dogs.txt, 在第一个文件中至少存储三只猫的名字, 在第二个文件中至少存储三条狗的名字. 编写一个程序, 尝试读取这些文件, 并将其内容打印到 ...
oracle对数据块的存取,简单了解数据在Oracle文件中的存储
2010-02-13 目的: 1. 了解数据块转储 2. 简单认证数据在ORACLE文件的存储测试环境: VM+Win2003+Oracle11g DB_BLOCK_SIZE 8k C:\Docum ...
简单了解数据在Oracle文件中的存储
2010-02-13 目的: 1. 了解数据块转储 2. 简单认证数据在ORACLE文件的存储测试环境: VM+Win2003+Oracle11g DB_BLOCK_SIZE 8k C:\Docum ...
深入剖析 RocketMQ 源码 - 消息存储模块
作者:vivo互联网服务器团队-Zhang Zhenglin 一.简介 RocketMQ 是阿里巴巴开源的分布式消息中间件,它借鉴了 Kafka 实现,支持消息订阅与发布.顺序消息.事务消息.定时消息 ...
kafka是如何通过offset定位一条消息的？
Kafka文件结构 Kafka 中消息是以 topic 进行分类的,生产者生产消息,消费者消费消息,都是面向 topic 的.topic 是逻辑上的概念,而 partition 是物理上的概念,每个 ...
如何让 GoldenGate 在 trail 文件中记录数据库的 SCN 信息
如何让 GoldenGate 在 trail 文件中记录数据库的 SCN 信息通常情况下,Oracle GoldenGate 的 trail 文件中是不会包含数据库的 SCN 信息的,要在 trai ...
python从txt拿取数据_python requests + xpath 获取分页详情页数据存入到txt文件中
直接代码,如有不懂请加群讨论 # *-* coding:utf-8 *-* # import json import requests import pytesseract import time i ...
CompactExifLib：访问JPEG文件中的EXIF标签
目录介绍背景演示应用程序使用代码读写标签标签ID和图像文件目录(IFD) 标签类型整数数组标签字符串有理数日期和时间原始数据和字节顺序移除标签加载和保存EXIF数据 GPS ...
python前n项和存为一个数组_python-将数组或DataFrame与其他信息一起保存在文件中...
python-将数组或DataFrame与其他信息一起保存在文件中统计软件Stata允许将短文本片段保存在数据集中. 这可以使用NumPy和/或NumPy完成. 这对我来说是非常有价值的功能,因为它 ...
include的头文件中include其他头文件时的路径问题
问题由于之前项目做得很少,一直没有注意到这个问题,include一直认为就是直接把include的文件复制过来,并没有去深究里面的原理.但今天做项目时发现如果是直接复制过来,那include的文件里 ...

RocketMq之一条消息在commitlog文件中如何存储验证

RocketMq之一条消息在commitlog文件中如何存储验证相关推荐

最新文章

热门文章