Java的RocketMQ之消息存儲(chǔ)和查詢?cè)碓斀?/h1>

 更新時(shí)間：2024年01月08日 09:31:13   作者：獵戶星座。  

這篇文章主要介紹了Java的RocketMQ之消息存儲(chǔ)和查詢?cè)碓斀?一臺(tái)Broker服務(wù)器只有一個(gè)CommitLog文件(組),RocketMQ會(huì)將所有主題的消息存儲(chǔ)在同一個(gè)文件中,這個(gè)文件中就存儲(chǔ)著一條條Message,每條Message都會(huì)按照順序?qū)懭?需要的朋友可以參考下

目錄

• 一、CommitLog
• 1、消息發(fā)送
• 2、讀取文件內(nèi)容
• 3、解析
• 4、輸出消息內(nèi)容
• 二、ConsumeQueue
• 1、MessageQueue
• 2、ConsumeQueue
• 3、解析文件
• 4、查詢消息
• 5、消費(fèi)消息
• 6、通過 Message Id 查詢
• 三、Index
• 1、index索引文件
• 2、構(gòu)建索引

一、CommitLog

CommitLog，消息存儲(chǔ)文件，所有主題的消息都存儲(chǔ)在 CommitLog 文件中。

我們的業(yè)務(wù)系統(tǒng)向 RocketMQ 發(fā)送一條消息，不管在中間經(jīng)歷了多么復(fù)雜的流程，最終這條消息會(huì)被持久化到CommitLog文件。

我們知道，一臺(tái)Broker服務(wù)器只有一個(gè)CommitLog文件(組)，RocketMQ會(huì)將所有主題的消息存儲(chǔ)在同一個(gè)文件中，這個(gè)文件中就存儲(chǔ)著一條條Message，每條Message都會(huì)按照順序?qū)懭搿?/p>

也許有時(shí)候，你會(huì)希望看看這個(gè) CommitLog 文件中，存儲(chǔ)的內(nèi)容到底長(zhǎng)什么樣子？

1、消息發(fā)送

當(dāng)然，我們需要先往 CommitLog 文件中寫入一些內(nèi)容，所以先來看一個(gè)消息發(fā)送的例子。

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個(gè)不同的主題中發(fā)送消息，如果只有一臺(tái)Broker機(jī)器，它們會(huì)保存到同一個(gè)CommitLog文件中。此時(shí)，這個(gè)文件的位置處于 C:/Users/shiqizhen/store/commitlog/00000000000000000000。

2、讀取文件內(nèi)容

這個(gè)文件我們不能直接打開，因?yàn)樗且粋€(gè)二進(jìn)制文件，所以我們需要通過程序來讀取它的字節(jié)數(shù)組。

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;
}

如上代碼，可以通過傳入文件的路徑，讀取該文件所有的內(nèi)容。為了方便下一步操作，我們把讀取到的字節(jié)數(shù)組轉(zhuǎn)換為java.nio.ByteBuffer對(duì)象。

3、解析

在解析之前，我們需要弄明白兩件事：

• 消息的格式，即一條消息包含哪些字段；
• 每個(gè)字段所占的字節(jié)大小。

在上面的圖中，我們已經(jīng)看到了消息的格式，包含了19個(gè)字段。關(guān)于字節(jié)大小，有的是 4 字節(jié)，有的是 8 字節(jié)，我們不再一一贅述，直接看代碼。

/**
 * 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、輸出消息內(nèi)容

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())+"隊(duì)列ID:"+ms.getQueueId()+" 存儲(chǔ)地址:"+ms.getStoreHost());
    }
}

運(yùn)行這段代碼，我們就可以直接看到CommitLog文件中的內(nèi)容：

主題:topic0 消息:清幽之地的博客 隊(duì)列ID:1 存儲(chǔ)地址:/192.168.44.1:10911
主題:topic1 消息:清幽之地的博客 隊(duì)列ID:0 存儲(chǔ)地址:/192.168.44.1:10911
主題:topic2 消息:清幽之地的博客 隊(duì)列ID:1 存儲(chǔ)地址:/192.168.44.1:10911
主題:topic3 消息:清幽之地的博客 隊(duì)列ID:0 存儲(chǔ)地址:/192.168.44.1:10911
主題:topic4 消息:清幽之地的博客 隊(duì)列ID:3 存儲(chǔ)地址:/192.168.44.1:10911
主題:topic5 消息:清幽之地的博客 隊(duì)列ID:1 存儲(chǔ)地址:/192.168.44.1:10911
主題:topic6 消息:清幽之地的博客 隊(duì)列ID:2 存儲(chǔ)地址:/192.168.44.1:10911
主題:topic7 消息:清幽之地的博客 隊(duì)列ID:3 存儲(chǔ)地址:/192.168.44.1:10911
主題:topic8 消息:清幽之地的博客 隊(duì)列ID:2 存儲(chǔ)地址:/192.168.44.1:10911
主題:topic9 消息:清幽之地的博客 隊(duì)列ID:0 存儲(chǔ)地址:/192.168.44.1:10911

不用過多的文字描述，通過上面這些代碼，相信你對(duì)CommitLog文件就有了更進(jìn)一步的了解。

此時(shí)，我們?cè)倏紤]另外一個(gè)問題：

CommitLog 文件保存了所有主題的消息，但我們消費(fèi)時(shí)，更多的是訂閱某一個(gè)主題進(jìn)行消費(fèi)。RocketMQ是怎么樣進(jìn)行高效的檢索消息的呢 ？

二、ConsumeQueue

為了解決上面那個(gè)問題，RocketMQ引入了ConsumeQueue消費(fèi)隊(duì)列文件。

在繼續(xù)往下說ConsumeQueue之前，我們必須先了解到另外一個(gè)概念，即MessageQueue。

1、MessageQueue

我們知道，在發(fā)送消息的時(shí)候，要指定一個(gè)Topic。那么，在創(chuàng)建Topic的時(shí)候，有一個(gè)很重要的參數(shù)MessageQueue。簡(jiǎn)單來說，就是你這個(gè)Topic對(duì)應(yīng)了多少個(gè)隊(duì)列，也就是幾個(gè)MessageQueue，默認(rèn)是4個(gè)。那它的作用是什么呢 ？

它是一個(gè)數(shù)據(jù)分片的機(jī)制。比如我們的Topic里面有100條數(shù)據(jù)，該Topic默認(rèn)是4個(gè)隊(duì)列，那么每個(gè)隊(duì)列中大約25條數(shù)據(jù)。 然后，這些MessageQueue是和Broker綁定在一起的，就是說每個(gè)MessageQueue都可能處于不同的Broker機(jī)器上，這取決于你的隊(duì)列數(shù)量和Broker集群。

我們來看上面的圖片，Topic名稱為order的主題，一共有4個(gè)MessageQueue，每個(gè)里面都有25條數(shù)據(jù)。因?yàn)樵诠P者的本地環(huán)境只有一個(gè)Broker，所以它們的brokerName都是指向同一臺(tái)機(jī)器。

既然MessageQueue是多個(gè)，那么在消息發(fā)送的時(shí)候，勢(shì)必要通過某種方式選擇一個(gè)隊(duì)列。默認(rèn)的情況下，就是通過輪詢來獲取一個(gè)消息隊(duì)列。

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);
}

當(dāng)然，RocketMQ還有一個(gè)故障延遲機(jī)制，在選擇消息隊(duì)列的時(shí)候會(huì)復(fù)雜一些，我們今天先不討論。

2、ConsumeQueue

說完了MessageQueue，我們接著來看ConsumerQueue。上面我們說，它是為了高效檢索主題消息的。

ConsumerQueue也是一組組文件，它的位置在C:/Users/shiqizhen/store/consumequeue。該目錄下面是以Topic命名的文件夾，然后再下一級(jí)是以MessageQueue隊(duì)列ID命名的文件夾，最后才是一個(gè)或多個(gè)文件。

這樣分層之后，RocketMQ至少可以得到以下幾個(gè)訊息：

• 先通過主題名稱，可以定位到具體的文件夾；
• 然后根據(jù)消息隊(duì)列ID找到具體的文件；
• 最后根據(jù)文件內(nèi)容，找到具體的消息。

那么，這個(gè)文件里面存儲(chǔ)的又是什么內(nèi)容呢 ？

3、解析文件

為了加速ConsumerQueue的檢索速度和節(jié)省磁盤空間，文件中不會(huì)存儲(chǔ)消息的全量消息。其存儲(chǔ)的格式如下：

同樣的，我們先寫一段代碼，按照這個(gè)格式輸出一下ConsumerQueue文件的內(nèi)容。

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("消息長(zhǎng)度:"+size+" 消息偏移量:" +offset);
    }
    System.out.println("--------------------------");
}

在前面，我們已經(jīng)向order這個(gè)主題中寫了100條數(shù)據(jù)，所以在這里它的order#messagequeue#0里面有25條記錄。

消息長(zhǎng)度:173 消息偏移量:2003
消息長(zhǎng)度:173 消息偏移量:2695
消息長(zhǎng)度:173 消息偏移量:3387
消息長(zhǎng)度:173 消息偏移量:4079
消息長(zhǎng)度:173 消息偏移量:4771
消息長(zhǎng)度:173 消息偏移量:5463
消息長(zhǎng)度:173 消息偏移量:6155
消息長(zhǎng)度:173 消息偏移量:6847
消息長(zhǎng)度:173 消息偏移量:7539
消息長(zhǎng)度:173 消息偏移量:8231
消息長(zhǎng)度:173 消息偏移量:8923
消息長(zhǎng)度:173 消息偏移量:9615
消息長(zhǎng)度:173 消息偏移量:10307
消息長(zhǎng)度:173 消息偏移量:10999
消息長(zhǎng)度:173 消息偏移量:11691
消息長(zhǎng)度:173 消息偏移量:12383
消息長(zhǎng)度:173 消息偏移量:13075
消息長(zhǎng)度:173 消息偏移量:13767
消息長(zhǎng)度:173 消息偏移量:14459
消息長(zhǎng)度:173 消息偏移量:15151
消息長(zhǎng)度:173 消息偏移量:15843
消息長(zhǎng)度:173 消息偏移量:16535
消息長(zhǎng)度:173 消息偏移量:17227
消息長(zhǎng)度:173 消息偏移量:17919
消息長(zhǎng)度:173 消息偏移量:18611

細(xì)心的朋友，肯定發(fā)現(xiàn)了。上面輸出的結(jié)果中，消息偏移量的差值等于 = 消息長(zhǎng)度 * 隊(duì)列長(zhǎng)度。

4、查詢消息

現(xiàn)在我們通過ConsumerQueue已經(jīng)知道了消息的長(zhǎng)度和偏移量，那么查找消息就比較容易了。

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;
}

然后，我們可以依靠這種方法，來實(shí)現(xiàn)通過ConsumerQueue獲取消息的具體內(nèi)容。

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文件內(nèi)容
	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文件內(nèi)容
				ByteBuffer buffer = CommitLogTest.read(path);
				while (true){
					//讀取消息偏移量和消息長(zhǎng)度
					long offset = (int) buffer.getLong();
					int size = buffer.getInt();
					long code = buffer.getLong();
					if (size==0){
						break;
					}
					//根據(jù)偏移量和消息長(zhǎng)度，在commitloh文件中讀取消息內(nèi)容
					MessageExt message = getMessageByOffset(commitLogBuffer,offset,size);
					if (message!=null){
						System.out.println("消息主題:"+message.getTopic()+" MessageQueue:"+
							message.getQueueId()+" 消息體:"+new String(message.getBody()));
					}
				}
			}
		}
	}
}

運(yùn)行這段代碼，就可以得到之前測(cè)試樣例中，10個(gè)主題的所有消息。

消息主題: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 消息體:清幽之地的博客

5、消費(fèi)消息

消息消費(fèi)的時(shí)候，其查找消息的過程也是差不多的。不過值得注意的一點(diǎn)是，ConsumerQueue文件和CommitLog文件可能都是多個(gè)的，所以會(huì)有一個(gè)定位文件的過程，我們來看源碼。

首先，根據(jù)消費(fèi)進(jìn)度來查找對(duì)應(yīng)的ConsumerQueue，獲取其文件內(nèi)容。

public SelectMappedBufferResult getIndexBuffer(final long startIndex) {
    //ConsumerQueue文件大小
    int mappedFileSize = this.mappedFileSize;	
    //根據(jù)消費(fèi)進(jìn)度，找到在consumerqueue文件里的偏移量
    long offset = startIndex * CQ_STORE_UNIT_SIZE;
    if (offset >= this.getMinLogicOffset()) {
        //返回ConsumerQueue映射文件
        MappedFile mappedFile = this.mappedFileQueue.findMappedFileByOffset(offset);
        if (mappedFile != null) {
            //返回文件里的某一塊內(nèi)容
            SelectMappedBufferResult result = mappedFile.selectMappedBuffer((int) (offset % mappedFileSize));
            return result;
        }
    }
    return null;
}

然后拿到消息在CommitLog文件中的偏移量和消息長(zhǎng)度，獲取消息。

public SelectMappedBufferResult getMessage(final long offset, final int size) {
    //commitlog文件大小
    int mappedFileSize = this.defaultMessageStore.getMessageStoreConfig().getMappedFileSizeCommitLog();
    //根據(jù)消息偏移量，定位到具體的commitlog文件
    MappedFile mappedFile = this.mappedFileQueue.findMappedFileByOffset(offset, offset == 0);
    if (mappedFile != null) {
        //根據(jù)消息偏移量和長(zhǎng)度，獲取消息內(nèi)容
        int pos = (int) (offset % mappedFileSize);
        return mappedFile.selectMappedBuffer(pos, size);
    }
    return null;
}

6、通過 Message Id 查詢

上面我們看到了通過消息偏移量來查找消息的方式，但RocketMQ還提供了其他幾種方式可以查詢消息。

• 通過Message Key 查詢；
• 通過Unique Key查詢；
• 通過Message Id查詢。

在這里，Message Key和Unique Key都是在消息發(fā)送之前，由客戶端生成的。我們可以自己設(shè)置，也可以由客戶端自動(dòng)生成，Message Id是在Broker端存儲(chǔ)消息的時(shí)候生成。

Message Id總共 16 字節(jié)，包含消息存儲(chǔ)主機(jī)地址和在CommitLog文件中的偏移量offset。有源碼為證：

/**
 * 創(chuàng)建消息ID
 * @param input     
 * @param addr      Broker服務(wù)器地址
 * @param offset    正在存儲(chǔ)的消息，在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());
}

當(dāng)我們根據(jù)Message Id向Broker查詢消息時(shí)，首先會(huì)通過一個(gè)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查詢消息的時(shí)候，實(shí)際上還是直接從特定Broker上的CommitLog指定位置進(jìn)行查詢，屬于精確查詢。

這個(gè)也沒問題，但是如果通過 Message Key 和 Unique Key 查詢的時(shí)候，RocketMQ 又是怎么做的呢？

三、Index

1、index索引文件

ConsumerQueue消息消費(fèi)隊(duì)列是專門為消息訂閱構(gòu)建的索引文件，提高根據(jù)主題與消息隊(duì)列檢索消息的速度。

另外，RocketMQ引入Hash索引機(jī)制，為消息建立索引，它的鍵就是Message Key 和 Unique Key。

那么，我們先看看index索引文件的結(jié)構(gòu)：

為了便于理解，我們還是以代碼的方式，來解析這個(gè)文件。

public static void main(String[] args) throws Exception {
    //index索引文件的路徑
    String path = "C:\\Users\\shiqizhen\\store\\index\\20200506224547616";
    ByteBuffer buffer = CommitLogTest.read(path);
    //該索引文件中包含消息的最小存儲(chǔ)時(shí)間
    long beginTimestamp = buffer.getLong();
    //該索引文件中包含消息的最大存儲(chǔ)時(shí)間
    long endTimestamp = buffer.getLong();
    //該索引文件中包含消息的最大物理偏移量(commitlog文件偏移量)
    long beginPhyOffset = buffer.getLong();
    //該索引文件中包含消息的最大物理偏移量(commitlog文件偏移量)
    long endPhyOffset = buffer.getLong();
    //hashslot個(gè)數(shù)
    int hashSlotCount = buffer.getInt();
    //Index條目列表當(dāng)前已使用的個(gè)數(shù)
    int indexCount = buffer.getInt();
    //500萬個(gè)hash槽，每個(gè)槽占4個(gè)字節(jié)，存儲(chǔ)的是index索引
    for (int i=0;i<5000000;i++){
        buffer.getInt();
    }
    //2000萬個(gè)index條目
    for (int j=0;j<20000000;j++){
        //消息key的hashcode
        int hashcode = buffer.getInt();
        //消息對(duì)應(yīng)的偏移量
        long offset = buffer.getLong();
        //消息存儲(chǔ)時(shí)間和第一條消息的差值
        int timedif = buffer.getInt();
        //該條目的上一條記錄的index索引
        int pre_no = buffer.getInt();
    }
    System.out.println(buffer.position()==buffer.capacity());
}

我們看最后輸出的結(jié)果為true，則證明解析的過程無誤。

2、構(gòu)建索引

我們發(fā)送的消息體中，包含Message Key 或 Unique Key，那么就會(huì)給它們每一個(gè)都構(gòu)建索引。

這里重點(diǎn)有兩個(gè)：

• 根據(jù)消息Key計(jì)算Hash槽的位置；
• 根據(jù)Hash槽的數(shù)量和Index索引來計(jì)算Index條目的起始位置。

將當(dāng)前 Index條目 的索引值，寫在Hash槽absSlotPos位置上；將Index條目的具體信息(hashcode/消息偏移量/時(shí)間差值/hash槽的值)，從起始偏移量absIndexPos開始，順序按字節(jié)寫入。

public boolean putKey(final String key, final long phyOffset, final long storeTimestamp) {
    if (this.indexHeader.getIndexCount() < this.indexNum) {
        //計(jì)算key的hash
        int keyHash = indexKeyHashMethod(key);
        //計(jì)算hash槽的坐標(biāo)
        int slotPos = keyHash % this.hashSlotNum;
        int absSlotPos = IndexHeader.INDEX_HEADER_SIZE + slotPos * hashSlotSize;
        //計(jì)算時(shí)間差值
        long timeDiff = storeTimestamp - this.indexHeader.getBeginTimestamp();
        timeDiff = timeDiff / 1000;
        //計(jì)算INDEX條目的起始偏移量
        int absIndexPos =
            IndexHeader.INDEX_HEADER_SIZE + this.hashSlotNum * hashSlotSize
                + this.indexHeader.getIndexCount() * indexSize;
        //依次寫入hashcode、消息偏移量、時(shí)間戳、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);
        //將當(dāng)前INDEX中包含的條目數(shù)量寫入HASH槽
        this.mappedByteBuffer.putInt(absSlotPos, this.indexHeader.getIndexCount());
        return true;
    }
    return false;
}

這樣構(gòu)建完Index索引之后，根據(jù)Message Key 或 Unique Key查詢消息就簡(jiǎn)單了。

比如我們通過RocketMQ客戶端工具，根據(jù)Unique Key來查詢消息。

adminImpl.queryMessageByUniqKey("order", "FD88E3AB24F6980059FDC9C3620464741BCC18B4AAC220FDFE890007");

在Broker端，通過Unique Key來計(jì)算Hash槽的位置，從而找到Index索引數(shù)據(jù)。從Index索引中拿到消息的物理偏移量，最后根據(jù)消息物理偏移量，直接到CommitLog文件中去找就可以了。

到此這篇關(guān)于Java的RocketMQ之消息存儲(chǔ)和查詢?cè)碓斀獾奈恼戮徒榻B到這了,更多相關(guān)RocketMQ消息存儲(chǔ)和查詢?cè)韮?nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章:

• RocketMQ事務(wù)消息使用與原理詳解
• RocketMQ順序消息的原理與特點(diǎn)
• RocketMQ事務(wù)消息原理與使用詳解
• RocketMQ消息重試機(jī)制原理分析講解
• RocketMQ線程池創(chuàng)建實(shí)現(xiàn)原理詳解
• RocketMQ生產(chǎn)者調(diào)用start發(fā)送消息原理示例

• RocketMQ
• 消息存儲(chǔ)
• 查詢
• 原理

相關(guān)文章

• 

  Java中RabbitMQ隊(duì)列實(shí)現(xiàn)RPC詳解

  這篇文章主要介紹了Java中RabbitMQ隊(duì)列實(shí)現(xiàn)RPC詳解,在本教程中,我們將使用RabbitMQ構(gòu)建一個(gè)RPC系統(tǒng)：一個(gè)客戶端和一個(gè)RPC服務(wù)器,我們將創(chuàng)建一個(gè)返回斐波那契數(shù)字的模擬RPC服務(wù),,需要的朋友可以參考下

  2023-08-08
• 

  java生成在線驗(yàn)證碼

  這篇文章主要介紹了java生成在線驗(yàn)證碼,需要的朋友可以參考下

  2023-10-10
• 

  Java的wait(), notify()和notifyAll()使用心得

  本篇文章是對(duì)java的 wait(),notify(),notifyAll()進(jìn)行了詳細(xì)的分析介紹，需要的朋友參考下

  2013-08-08
• 

  SpringBoot整合Lombok插件與使用詳解

  Lombok是Java開發(fā)的插件,通過注解自動(dòng)生成常用代碼,如getter/setter,節(jié)省開發(fā)時(shí)間,提高效率,它在編譯期生成方法,不影響性能,安裝Lombok需要添加Maven依賴和IDEA插件,使用注解如@Data、@Getter等簡(jiǎn)化代碼編寫,官網(wǎng)提供詳細(xì)文檔

  2024-09-09
• 

  Java類型通配符應(yīng)用實(shí)戰(zhàn)分析

  這篇文章主要介紹了Java類型通配符應(yīng)用實(shí)戰(zhàn),簡(jiǎn)單分析了Java類型通配符概念、原理并結(jié)合實(shí)例形式給出了Java類型通配符相關(guān)使用技巧,需要的朋友可以參考下

  2019-07-07
• 

  spring session同域下單點(diǎn)登錄實(shí)現(xiàn)解析

  這篇文章主要介紹了spring session同域下單點(diǎn)登錄實(shí)現(xiàn)解析,文中通過示例代碼介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友可以參考下

  2019-10-10
• 

  springboot使用Thymeleaf報(bào)錯(cuò)常見的幾種解決方案

  這篇文章主要介紹了springboot使用Thymeleaf報(bào)錯(cuò)常見的幾種解決方案，具有很好的參考價(jià)值，希望對(duì)大家有所幫助。如有錯(cuò)誤或未考慮完全的地方，望不吝賜教

  2022-11-11
• 

  JSON,AJAX,Maven入門基礎(chǔ)

  這篇文章主要介紹了JSON,AJAX和Maven基礎(chǔ)，如何使用AJAX讀取Json數(shù)組里面的數(shù)據(jù)，感興趣的小伙伴們可以參考一下，希望能夠幫助到你

  2021-07-07
• 

  SpringSecurity整合jwt權(quán)限認(rèn)證的全流程講解

  這篇文章主要介紹了SpringSecurity整合jwt權(quán)限認(rèn)證的全流程講解，具有很好的參考價(jià)值，希望對(duì)大家有所幫助。如有錯(cuò)誤或未考慮完全的地方，望不吝賜教

  2021-06-06
• 

  輕松掌握java組合模式

  這篇文章主要幫助大家輕松掌握java組合模式,具有一定的參考價(jià)值，感興趣的小伙伴們可以參考一下

  2016-09-09

最新評(píng)論