RocketMQ 作為一款優秀的分布式消息中間件，可以為業務方提供高性能低延遲的穩定可靠的消息服務。其核心優勢是可靠的消費存儲、消息發送的高性能和低延遲、強大的消息堆積能力和消息處理能力。
從存儲方式來看，主要有幾個方面：
- 文件系統
- 分布式KV存儲
- 關系型數據庫
從效率上來講，文件系統高於KV存儲，KV存儲又高於關系型數據庫。因為直接操作文件系統肯定是最快的，那么業界主流的消息隊列中間件，如RocketMQ 、RabbitMQ 、kafka 都是采用文件系統的方式來存儲消息。
今天，我們就從它的存儲文件入手，來探索一下 RocketMQ 消息存儲的機制。
一、CommitLog
CommitLog，消息存儲文件，所有主題的消息都存儲在 CommitLog 文件中。
我們的業務系統向 RocketMQ 發送一條消息，不管在中間經歷了多么復雜的流程，最終這條消息會被持久化到CommitLog文件。
我們知道，一台Broker服務器只有一個CommitLog文件(組)，RocketMQ會將所有主題的消息存儲在同一個文件中，這個文件中就存儲着一條條Message，每條Message都會按照順序寫入。
也許有時候，你會希望看看這個 CommitLog 文件中，存儲的內容到底長什么樣子？
1、消息發送
當然，我們需要先往 CommitLog 文件中寫入一些內容，所以先來看一個消息發送的例子。
public static void main(String[] args) throws Exception { MQProducer producer = getProducer(); for (int i = 0;i<10;i++){ Message message = new Message(); message.setTopic("topic"+i); message.setBody(("清幽之地的博客").getBytes()); SendResult sendResult = producer.send(message); } producer.shutdown(); }
我們向10個不同的主題中發送消息，如果只有一台Broker機器，它們會保存到同一個CommitLog文件中。此時，這個文件的位置處於 C:/Users/shiqizhen/store/commitlog/00000000000000000000。
2、讀取文件內容
這個文件我們不能直接打開，因為它是一個二進制文件，所以我們需要通過程序來讀取它的字節數組。
public static ByteBuffer read(String path)throws Exception{ File file = new File(path); FileInputStream fin = new FileInputStream(file); byte[] bytes = new byte[(int)file.length()]; fin.read(bytes); ByteBuffer buffer = ByteBuffer.wrap(bytes); return buffer; }
如上代碼，可以通過傳入文件的路徑，讀取該文件所有的內容。為了方便下一步操作，我們把讀取到的字節數組轉換為java.nio.ByteBuffer對象。
3、解析
在解析之前，我們需要弄明白兩件事：
- 消息的格式，即一條消息包含哪些字段；
- 每個字段所占的字節大小。
在上面的圖中，我們已經看到了消息的格式，包含了19個字段。關於字節大小，有的是 4 字節，有的是 8 字節，我們不再一一贅述，直接看代碼。
/** * commitlog 文件解析 * @param byteBuffer * @return * @throws Exception */ public static MessageExt decodeCommitLog(ByteBuffer byteBuffer)throws Exception { MessageExt msgExt = new MessageExt(); // 1 TOTALSIZE int storeSize = byteBuffer.getInt(); msgExt.setStoreSize(storeSize); if (storeSize<=0){ return null; } // 2 MAGICCODE byteBuffer.getInt(); // 3 BODYCRC int bodyCRC = byteBuffer.getInt(); msgExt.setBodyCRC(bodyCRC); // 4 QUEUEID int queueId = byteBuffer.getInt(); msgExt.setQueueId(queueId); // 5 FLAG int flag = byteBuffer.getInt(); msgExt.setFlag(flag); // 6 QUEUEOFFSET long queueOffset = byteBuffer.getLong(); msgExt.setQueueOffset(queueOffset); // 7 PHYSICALOFFSET long physicOffset = byteBuffer.getLong(); msgExt.setCommitLogOffset(physicOffset); // 8 SYSFLAG int sysFlag = byteBuffer.getInt(); msgExt.setSysFlag(sysFlag); // 9 BORNTIMESTAMP long bornTimeStamp = byteBuffer.getLong(); msgExt.setBornTimestamp(bornTimeStamp); // 10 BORNHOST int bornhostIPLength = (sysFlag & MessageSysFlag.BORNHOST_V6_FLAG) == 0 ? 4 : 16; byte[] bornHost = new byte[bornhostIPLength]; byteBuffer.get(bornHost, 0, bornhostIPLength); int port = byteBuffer.getInt(); msgExt.setBornHost(new InetSocketAddress(InetAddress.getByAddress(bornHost), port)); // 11 STORETIMESTAMP long storeTimestamp = byteBuffer.getLong(); msgExt.setStoreTimestamp(storeTimestamp); // 12 STOREHOST int storehostIPLength = (sysFlag & MessageSysFlag.STOREHOSTADDRESS_V6_FLAG) == 0 ? 4 : 16; byte[] storeHost = new byte[storehostIPLength]; byteBuffer.get(storeHost, 0, storehostIPLength); port = byteBuffer.getInt(); msgExt.setStoreHost(new InetSocketAddress(InetAddress.getByAddress(storeHost), port)); // 13 RECONSUMETIMES int reconsumeTimes = byteBuffer.getInt(); msgExt.setReconsumeTimes(reconsumeTimes); // 14 Prepared Transaction Offset long preparedTransactionOffset = byteBuffer.getLong(); msgExt.setPreparedTransactionOffset(preparedTransactionOffset); // 15 BODY int bodyLen = byteBuffer.getInt(); if (bodyLen > 0) { byte[] body = new byte[bodyLen]; byteBuffer.get(body); msgExt.setBody(body); } // 16 TOPIC byte topicLen = byteBuffer.get(); byte[] topic = new byte[(int) topicLen]; byteBuffer.get(topic); msgExt.setTopic(new String(topic, CHARSET_UTF8)); // 17 properties short propertiesLength = byteBuffer.getShort(); if (propertiesLength > 0) { byte[] properties = new byte[propertiesLength]; byteBuffer.get(properties); String propertiesString = new String(properties, CHARSET_UTF8); Map<String, String> map = string2messageProperties(propertiesString); } int msgIDLength = storehostIPLength + 4 + 8; ByteBuffer byteBufferMsgId = ByteBuffer.allocate(msgIDLength); String msgId = createMessageId(byteBufferMsgId, msgExt.getStoreHostBytes(), msgExt.getCommitLogOffset()); msgExt.setMsgId(msgId); return msgExt; }
4、輸出消息內容
public static void main(String[] args) throws Exception { String filePath = "C:\\Users\\shiqizhen\\store\\commitlog\\00000000000000000000"; ByteBuffer buffer = read(filePath); List<MessageExt> messageList = new ArrayList<>(); while (true){ MessageExt message = decodeCommitLog(buffer); if (message==null){ break; } messageList.add(message); } for (MessageExt ms:messageList) { System.out.println("主題:"+ms.getTopic()+" 消息:"+ new String(ms.getBody())+"隊列ID:"+ms.getQueueId()+" 存儲地址:"+ms.getStoreHost()); } }
運行這段代碼，我們就可以直接看到CommitLog文件中的內容：
主題:topic0 消息:RocketMQ消息存儲和查詢實戰 隊列ID:1 存儲地址:/192.168.44.1:10911 主題:topic1 消息:RocketMQ消息存儲和查詢實戰 隊列ID:0 存儲地址:/192.168.44.1:10911 主題:topic2 消息:RocketMQ消息存儲和查詢實戰 隊列ID:1 存儲地址:/192.168.44.1:10911 主題:topic3 消息:RocketMQ消息存儲和查詢實戰 隊列ID:0 存儲地址:/192.168.44.1:10911 主題:topic4 消息:RocketMQ消息存儲和查詢實戰 隊列ID:3 存儲地址:/192.168.44.1:10911 主題:topic5 消息:RocketMQ消息存儲和查詢實戰 隊列ID:1 存儲地址:/192.168.44.1:10911 主題:topic6 消息:RocketMQ消息存儲和查詢實戰 隊列ID:2 存儲地址:/192.168.44.1:10911 主題:topic7 消息:RocketMQ消息存儲和查詢實戰 隊列ID:3 存儲地址:/192.168.44.1:10911 主題:topic8 消息:RocketMQ消息存儲和查詢實戰 隊列ID:2 存儲地址:/192.168.44.1:10911 主題:topic9 消息:RocketMQ消息存儲和查詢實戰 隊列ID:0 存儲地址:/192.168.44.1:10911
不用過多的文字描述，通過上面這些代碼，相信你對CommitLog文件就有了更進一步的了解。
此時，我們再考慮另外一個問題：
CommitLog 文件保存了所有主題的消息，但我們消費時，更多的是訂閱某一個主題進行消費。RocketMQ是怎么樣進行高效的檢索消息的呢 ？
二、ConsumeQueue
為了解決上面那個問題，RocketMQ引入了ConsumeQueue消費隊列文件。
在繼續往下說ConsumeQueue之前，我們必須先了解到另外一個概念，即MessageQueue。
1、MessageQueue
我們知道，在發送消息的時候，要指定一個Topic。那么，在創建Topic的時候，有一個很重要的參數MessageQueue。簡單來說，就是你這個Topic對應了多少個隊列，也就是幾個MessageQueue，默認是4個。那它的作用是什么呢 ？
它是一個數據分片的機制。比如我們的Topic里面有100條數據，該Topic默認是4個隊列，那么每個隊列中大約25條數據。 然后，這些MessageQueue是和Broker綁定在一起的，就是說每個MessageQueue都可能處於不同的Broker機器上，這取決於你的隊列數量和Broker集群。
我們來看上面的圖片，Topic名稱為order的主題，一共有4個MessageQueue，每個里面都有25條數據。因為在筆者的本地環境只有一個Broker，所以它們的brokerName都是指向同一台機器。
既然MessageQueue是多個，那么在消息發送的時候，勢必要通過某種方式選擇一個隊列。默認的情況下，就是通過輪詢來獲取一個消息隊列。
public MessageQueue selectOneMessageQueue() { int index = this.sendWhichQueue.getAndIncrement(); int pos = Math.abs(index) % this.messageQueueList.size(); if (pos < 0) pos = 0; return this.messageQueueList.get(pos); }
當然，RocketMQ還有一個故障延遲機制，在選擇消息隊列的時候會復雜一些，我們今天先不討論。
2、ConsumeQueue
說完了MessageQueue，我們接着來看ConsumerQueue。上面我們說，它是為了高效檢索主題消息的。
ConsumerQueue也是一組組文件，它的位置在C:/Users/shiqizhen/store/consumequeue。該目錄下面是以Topic命名的文件夾，然后再下一級是以MessageQueue隊列ID命名的文件夾，最后才是一個或多個文件。
這樣分層之后，RocketMQ至少可以得到以下幾個訊息：
- 先通過主題名稱，可以定位到具體的文件夾；
- 然后根據消息隊列ID找到具體的文件；
- 最后根據文件內容，找到具體的消息。
那么，這個文件里面存儲的又是什么內容呢 ？
3、解析文件
為了加速ConsumerQueue的檢索速度和節省磁盤空間，文件中不會存儲消息的全量消息。其存儲的格式如下：
同樣的，我們先寫一段代碼，按照這個格式輸出一下ConsumerQueue文件的內容。
public static void main(String[] args)throws Exception { String path = "C:\\Users\\shiqizhen\\store\\consumequeue\\order\\0\\00000000000000000000"; ByteBuffer buffer = read(path); while (true){ long offset = buffer.getLong(); long size = buffer.getInt(); long code = buffer.getLong(); if (size==0){ break; } System.out.println("消息長度:"+size+" 消息偏移量:" +offset); } System.out.println("--------------------------"); }
在前面，我們已經向order這個主題中寫了100條數據，所以在這里它的order#messagequeue#0里面有25條記錄。
消息長度:173 消息偏移量:2003 消息長度:173 消息偏移量:2695 消息長度:173 消息偏移量:3387 消息長度:173 消息偏移量:4079 消息長度:173 消息偏移量:4771 消息長度:173 消息偏移量:5463 消息長度:173 消息偏移量:6155 消息長度:173 消息偏移量:6847 消息長度:173 消息偏移量:7539 消息長度:173 消息偏移量:8231 消息長度:173 消息偏移量:8923 消息長度:173 消息偏移量:9615 消息長度:173 消息偏移量:10307 消息長度:173 消息偏移量:10999 消息長度:173 消息偏移量:11691 消息長度:173 消息偏移量:12383 消息長度:173 消息偏移量:13075 消息長度:173 消息偏移量:13767 消息長度:173 消息偏移量:14459 消息長度:173 消息偏移量:15151 消息長度:173 消息偏移量:15843 消息長度:173 消息偏移量:16535 消息長度:173 消息偏移量:17227 消息長度:173 消息偏移量:17919 消息長度:173 消息偏移量:18611 --------------------------
細心的朋友，肯定發現了。上面輸出的結果中，消息偏移量的差值等於 = 消息長度 * 隊列長度。
4、查詢消息
現在我們通過ConsumerQueue已經知道了消息的長度和偏移量，那么查找消息就比較容易了。
public static MessageExt getMessageByOffset(ByteBuffer commitLog,long offset,int size) throws Exception { ByteBuffer slice = commitLog.slice(); slice.position((int)offset); slice.limit((int) (offset+size)); MessageExt message = CommitLogTest.decodeCommitLog(slice); return message; }
然后，我們可以依靠這種方法，來實現通過ConsumerQueue獲取消息的具體內容。
public static void main(String[] args) throws Exception { //consumerqueue根目錄 String consumerPath = "C:\\Users\\shiqizhen\\store\\consumequeue"; //commitlog目錄 String commitLogPath = "C:\\Users\\shiqizhen\\store\\commitlog\\00000000000000000000"; //讀取commitlog文件內容 ByteBuffer commitLogBuffer = CommitLogTest.read(commitLogPath); //遍歷consumerqueue目錄下的所有文件 File file = new File(consumerPath); File[] files = file.listFiles(); for (File f:files) { if (f.isDirectory()){ File[] listFiles = f.listFiles(); for (File queuePath:listFiles) { String path = queuePath+"/00000000000000000000"; //讀取consumerqueue文件內容 ByteBuffer buffer = CommitLogTest.read(path); while (true){ //讀取消息偏移量和消息長度 long offset = (int) buffer.getLong(); int size = buffer.getInt(); long code = buffer.getLong(); if (size==0){ break; } //根據偏移量和消息長度，在commitloh文件中讀取消息內容 MessageExt message = getMessageByOffset(commitLogBuffer,offset,size); if (message!=null){ System.out.println("消息主題:"+message.getTopic()+" MessageQueue:"+ message.getQueueId()+" 消息體:"+new String(message.getBody())); } } } } } }
運行這段代碼，就可以得到之前測試樣例中，10個主題的所有消息。
消息主題:topic0 MessageQueue:1 消息體:RocketMQ消息存儲和查詢實戰 消息主題:topic1 MessageQueue:0 消息體:RocketMQ消息存儲和查詢實戰 消息主題:topic2 MessageQueue:1 消息體:RocketMQ消息存儲和查詢實戰 消息主題:topic3 MessageQueue:0 消息體:RocketMQ消息存儲和查詢實戰 消息主題:topic4 MessageQueue:3 消息體:RocketMQ消息存儲和查詢實戰 消息主題:topic5 MessageQueue:1 消息體:RocketMQ消息存儲和查詢實戰 消息主題:topic6 MessageQueue:2 消息體:RocketMQ消息存儲和查詢實戰 消息主題:topic7 MessageQueue:3 消息體:RocketMQ消息存儲和查詢實戰 消息主題:topic8 MessageQueue:2 消息體:RocketMQ消息存儲和查詢實戰 消息主題:topic9 MessageQueue:0 消息體:RocketMQ消息存儲和查詢實戰
5、消費消息
消息消費的時候，其查找消息的過程也是差不多的。不過值得注意的一點是，ConsumerQueue文件和CommitLog文件可能都是多個的，所以會有一個定位文件的過程，我們來看源碼。
首先，根據消費進度來查找對應的ConsumerQueue，獲取其文件內容。
public SelectMappedBufferResult getIndexBuffer(final long startIndex) { //ConsumerQueue文件大小 int mappedFileSize = this.mappedFileSize; //根據消費進度，找到在consumerqueue文件里的偏移量 long offset = startIndex * CQ_STORE_UNIT_SIZE; if (offset >= this.getMinLogicOffset()) { //返回ConsumerQueue映射文件 MappedFile mappedFile = this.mappedFileQueue.findMappedFileByOffset(offset); if (mappedFile != null) { //返回文件里的某一塊內容 SelectMappedBufferResult result = mappedFile.selectMappedBuffer((int) (offset % mappedFileSize)); return result; } } return null; }
然后拿到消息在CommitLog文件中的偏移量和消息長度，獲取消息。
public SelectMappedBufferResult getMessage(final long offset, final int size) { //commitlog文件大小 int mappedFileSize = this.defaultMessageStore.getMessageStoreConfig().getMappedFileSizeCommitLog(); //根據消息偏移量，定位到具體的commitlog文件 MappedFile mappedFile = this.mappedFileQueue.findMappedFileByOffset(offset, offset == 0); if (mappedFile != null) { //根據消息偏移量和長度，獲取消息內容 int pos = (int) (offset % mappedFileSize); return mappedFile.selectMappedBuffer(pos, size); } return null; }
三、Index
上面我們看到了通過消息偏移量來查找消息的方式，但RocketMQ還提供了其他幾種方式可以查詢消息。
- 通過Message Key 查詢；
- 通過Unique Key查詢；
- 通過Message Id查詢。
在這里，Message Key和Unique Key都是在消息發送之前，由客戶端生成的。我們可以自己設置，也可以由客戶端自動生成，Message Id是在Broker端存儲消息的時候生成。
1、通過 Message Id 查詢
Message Id總共 16 字節，包含消息存儲主機地址和在CommitLog文件中的偏移量offset。有源碼為證：
/** * 創建消息ID * @param input * @param addr Broker服務器地址 * @param offset 正在存儲的消息，在Commitlog中的偏移量 * @return */ public static String createMessageId(final ByteBuffer input, final ByteBuffer addr, final long offset) { input.flip(); int msgIDLength = addr.limit() == 8 ? 16 : 28; input.limit(msgIDLength); input.put(addr); input.putLong(offset); return UtilAll.bytes2string(input.array()); }
當我們根據Message Id向Broker查詢消息時，首先會通過一個decodeMessageId方法，將Broker地址和消息的偏移量解析出來。
public static MessageId decodeMessageId(final String msgId) throws Exception { SocketAddress address; long offset; int ipLength = msgId.length() == 32 ? 4 * 2 : 16 * 2; byte[] ip = UtilAll.string2bytes(msgId.substring(0, ipLength)); byte[] port = UtilAll.string2bytes(msgId.substring(ipLength, ipLength + 8)); ByteBuffer bb = ByteBuffer.wrap(port); int portInt = bb.getInt(0); //解析出來Broker地址 address = new InetSocketAddress(InetAddress.getByAddress(ip), portInt); //偏移量 byte[] data = UtilAll.string2bytes(msgId.substring(ipLength + 8, ipLength + 8 + 16)); bb = ByteBuffer.wrap(data); offset = bb.getLong(0); return new MessageId(address, offset); }
所以通過Message Id查詢消息的時候，實際上還是直接從特定Broker上的CommitLog指定位置進行查詢，屬於精確查詢。
這個也沒問題，但是如果通過 Message Key 和 Unique Key 查詢的時候，RocketMQ 又是怎么做的呢？
2、index索引文件
ConsumerQueue消息消費隊列是專門為消息訂閱構建的索引文件，提高根據主題與消息隊列檢索消息的速度。
另外，RocketMQ引入Hash索引機制，為消息建立索引，它的鍵就是Message Key 和 Unique Key。
那么，我們先看看index索引文件的結構：
為了便於理解，我們還是以代碼的方式，來解析這個文件。
public static void main(String[] args) throws Exception { //index索引文件的路徑 String path = "C:\\Users\\shiqizhen\\store\\index\\20200506224547616"; ByteBuffer buffer = CommitLogTest.read(path); //該索引文件中包含消息的最小存儲時間 long beginTimestamp = buffer.getLong(); //該索引文件中包含消息的最大存儲時間 long endTimestamp = buffer.getLong(); //該索引文件中包含消息的最大物理偏移量(commitlog文件偏移量) long beginPhyOffset = buffer.getLong(); //該索引文件中包含消息的最大物理偏移量(commitlog文件偏移量) long endPhyOffset = buffer.getLong(); //hashslot個數 int hashSlotCount = buffer.getInt(); //Index條目列表當前已使用的個數 int indexCount = buffer.getInt(); //500萬個hash槽，每個槽占4個字節，存儲的是index索引 for (int i=0;i<5000000;i++){ buffer.getInt(); } //2000萬個index條目 for (int j=0;j<20000000;j++){ //消息key的hashcode int hashcode = buffer.getInt(); //消息對應的偏移量 long offset = buffer.getLong(); //消息存儲時間和第一條消息的差值 int timedif = buffer.getInt(); //該條目的上一條記錄的index索引 int pre_no = buffer.getInt(); } System.out.println(buffer.position()==buffer.capacity()); }
我們看最后輸出的結果為true，則證明解析的過程無誤。
3、構建索引
我們發送的消息體中，包含Message Key 或 Unique Key，那么就會給它們每一個都構建索引。
這里重點有兩個：
- 根據消息Key計算Hash槽的位置；
- 根據Hash槽的數量和Index索引來計算Index條目的起始位置。
將當前 Index條目 的索引值，寫在Hash槽absSlotPos位置上；將Index條目的具體信息(hashcode/消息偏移量/時間差值/hash槽的值)，從起始偏移量absIndexPos開始，順序按字節寫入。
public boolean putKey(final String key, final long phyOffset, final long storeTimestamp) { if (this.indexHeader.getIndexCount() < this.indexNum) { //計算key的hash int keyHash = indexKeyHashMethod(key); //計算hash槽的坐標 int slotPos = keyHash % this.hashSlotNum; int absSlotPos = IndexHeader.INDEX_HEADER_SIZE + slotPos * hashSlotSize; //計算時間差值 long timeDiff = storeTimestamp - this.indexHeader.getBeginTimestamp(); timeDiff = timeDiff / 1000; //計算INDEX條目的起始偏移量 int absIndexPos = IndexHeader.INDEX_HEADER_SIZE + this.hashSlotNum * hashSlotSize + this.indexHeader.getIndexCount() * indexSize; //依次寫入hashcode、消息偏移量、時間戳、hash槽的值 this.mappedByteBuffer.putInt(absIndexPos, keyHash); this.mappedByteBuffer.putLong(absIndexPos + 4, phyOffset); this.mappedByteBuffer.putInt(absIndexPos + 4 + 8, (int) timeDiff); this.mappedByteBuffer.putInt(absIndexPos + 4 + 8 + 4, slotValue); //將當前INDEX中包含的條目數量寫入HASH槽 this.mappedByteBuffer.putInt(absSlotPos, this.indexHeader.getIndexCount()); return true; } return false; }
這樣構建完Index索引之后，根據Message Key 或 Unique Key查詢消息就簡單了。
比如我們通過RocketMQ客戶端工具，根據Unique Key來查詢消息。
adminImpl.queryMessageByUniqKey("order", "FD88E3AB24F6980059FDC9C3620464741BCC18B4AAC220FDFE890007");
在Broker端，通過Unique Key來計算Hash槽的位置，從而找到Index索引數據。從Index索引中拿到消息的物理偏移量，最后根據消息物理偏移量，直接到CommitLog文件中去找就可以了。