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）：1166
Process finished with exit code 0
总结
本次主要根据《RocketMQ技术内幕》这本书为依据，对存储在commitLOg文件中的内容进行了探索。可以从控制台打印中，明显感觉到与RocketMQ设计的吻合。比如文件名字与解析后得到的消息物理存储偏移量一致。本文crc这种没有正确的解析，这种可以使用RocketMQ提供的工具进行解析。还有用到了糊涂的工具包，用来解析2进制到16进制。
作者：小安子antianchi
