RocketMQ消息存儲和查詢原理

前言

RocketMQ 做爲一款優秀的分佈式消息中間件,能夠爲業務方提供高性能低延遲的穩定可靠的消息服務。其核心優點是可靠的消費存儲、消息發送的高性能和低延遲、強大的消息堆積能力和消息處理能力。java

從存儲方式來看,主要有幾個方面:數據庫

  • 文件系統
  • 分佈式KV存儲
  • 關係型數據庫

從效率上來說,文件系統高於KV存儲,KV存儲又高於關係型數據庫。由於直接操做文件系統確定是最快的,那麼業界主流的消息隊列中間件,如RocketMQ 、RabbitMQ 、kafka 都是採用文件系統的方式來存儲消息。數組

今天,咱們就從它的存儲文件入手,來探索一下 RocketMQ 消息存儲的機制。bash

1、CommitLog

CommitLog,消息存儲文件,全部主題的消息都存儲在 CommitLog 文件中。服務器

咱們的業務系統向 RocketMQ 發送一條消息,無論在中間經歷了多麼複雜的流程,最終這條消息會被持久化到CommitLog文件。app

咱們知道,一臺Broker服務器只有一個CommitLog文件(組),RocketMQ會將全部主題的消息存儲在同一個文件中,這個文件中就存儲着一條條Message,每條Message都會按照順序寫入。 分佈式

也許有時候,你會但願看看這個 CommitLog 文件中,存儲的內容到底長什麼樣子?工具

一、消息發送

固然,咱們須要先往 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測試

二、讀取文件內容

這個文件咱們不能直接打開,由於它是一個二進制文件,因此咱們須要經過程序來讀取它的字節數組。

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對象。

三、解析

在解析以前,咱們須要弄明白兩件事:

  • 消息的格式,即一條消息包含哪些字段;
  • 每一個字段所佔的字節大小。

在上面的圖中,咱們已經看到了消息的格式,包含了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;
}
複製代碼

四、輸出消息內容

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 消息:清幽之地的博客 隊列ID:1 存儲地址:/192.168.44.1:10911
主題:topic1 消息:清幽之地的博客 隊列ID:0 存儲地址:/192.168.44.1:10911
主題:topic2 消息:清幽之地的博客 隊列ID:1 存儲地址:/192.168.44.1:10911
主題:topic3 消息:清幽之地的博客 隊列ID:0 存儲地址:/192.168.44.1:10911
主題:topic4 消息:清幽之地的博客 隊列ID:3 存儲地址:/192.168.44.1:10911
主題:topic5 消息:清幽之地的博客 隊列ID:1 存儲地址:/192.168.44.1:10911
主題:topic6 消息:清幽之地的博客 隊列ID:2 存儲地址:/192.168.44.1:10911
主題:topic7 消息:清幽之地的博客 隊列ID:3 存儲地址:/192.168.44.1:10911
主題:topic8 消息:清幽之地的博客 隊列ID:2 存儲地址:/192.168.44.1:10911
主題:topic9 消息:清幽之地的博客 隊列ID:0 存儲地址:/192.168.44.1:10911
複製代碼

不用過多的文字描述,經過上面這些代碼,相信你對CommitLog文件就有了更進一步的瞭解。

此時,咱們再考慮另一個問題:

CommitLog 文件保存了全部主題的消息,但咱們消費時,更多的是訂閱某一個主題進行消費。RocketMQ是怎麼樣進行高效的檢索消息的呢 ?

2、ConsumeQueue

爲了解決上面那個問題,RocketMQ引入了ConsumeQueue消費隊列文件。

在繼續往下說ConsumeQueue以前,咱們必須先了解到另一個概念,即MessageQueue

一、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還有一個故障延遲機制,在選擇消息隊列的時候會複雜一些,咱們今天先不討論。

二、ConsumeQueue

說完了MessageQueue,咱們接着來看ConsumerQueue。上面咱們說,它是爲了高效檢索主題消息的。

ConsumerQueue也是一組組文件,它的位置在C:/Users/shiqizhen/store/consumequeue。該目錄下面是以Topic命名的文件夾,而後再下一級是以MessageQueue隊列ID命名的文件夾,最後纔是一個或多個文件。

這樣分層以後,RocketMQ至少能夠獲得如下幾個訊息:

  • 先經過主題名稱,能夠定位到具體的文件夾;
  • 而後根據消息隊列ID找到具體的文件;
  • 最後根據文件內容,找到具體的消息。

那麼,這個文件裏面存儲的又是什麼內容呢 ?

三、解析文件

爲了加速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
--------------------------
複製代碼

細心的朋友,確定發現了。上面輸出的結果中,消息偏移量的差值等於 = 消息長度 * 隊列長度。

四、查詢消息

如今咱們經過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 消息體:清幽之地的博客
消息主題:topic1 MessageQueue:0 消息體:清幽之地的博客
消息主題:topic2 MessageQueue:1 消息體:清幽之地的博客
消息主題:topic3 MessageQueue:0 消息體:清幽之地的博客
消息主題:topic4 MessageQueue:3 消息體:清幽之地的博客
消息主題:topic5 MessageQueue:1 消息體:清幽之地的博客
消息主題:topic6 MessageQueue:2 消息體:清幽之地的博客
消息主題:topic7 MessageQueue:3 消息體:清幽之地的博客
消息主題:topic8 MessageQueue:2 消息體:清幽之地的博客
消息主題:topic9 MessageQueue:0 消息體:清幽之地的博客
複製代碼

五、消費消息

消息消費的時候,其查找消息的過程也是差很少的。不過值得注意的一點是,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;
}
複製代碼

六、經過 Message Id 查詢

上面咱們看到了經過消息偏移量來查找消息的方式,但RocketMQ還提供了其餘幾種方式能夠查詢消息。

  • 經過Message Key 查詢;
  • 經過Unique Key查詢;
  • 經過Message Id查詢。

在這裏,Message Key和Unique Key都是在消息發送以前,由客戶端生成的。咱們能夠本身設置,也能夠由客戶端自動生成,Message Id是在Broker端存儲消息的時候生成。

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 又是怎麼作的呢?

3、Index

一、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,則證實解析的過程無誤。

二、構建索引

咱們發送的消息體中,包含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文件中去找就能夠了。

總結

本文探討了RocketMQ中消息存儲和消息查找的基本思路。源碼中間過程都很複雜,可是經過這種自下而上的方式,直接從文件入手,剖析它們的文件結構,從而梳理清楚它們的關係和做用,但願能對朋友們產生積極做用。

原創不易,客官們點個贊再走嘛,這將是筆者持續寫做的動力~

相關文章
相關標籤/搜索