Shulou(Shulou.com)06/01 Report--

Editor to share with you how to store broker messages in RocketMQ to achieve message dump. I hope you will get something after reading this article. Let's discuss it together.

After receiving the message sent by producer, broker first stores the message at the end of CommitLog, and then dumps the message to ConsumeQueue and IndexFile through an asynchronous distribution thread, ReputMessageService.

The core logic of the dump is in ReputMessageService.doReput:

/ / DefaultMessageStore.ReputMessageService private void doReput () {if (this.reputFromOffset)

< DefaultMessageStore.this.commitLog.getMinOffset()) { log.warn("The reputFromOffset={} is smaller than minPyOffset={}, this usually indicate that the dispatch behind too much and the commitlog has expired.", this.reputFromOffset, DefaultMessageStore.this.commitLog.getMinOffset()); this.reputFromOffset = DefaultMessageStore.this.commitLog.getMinOffset(); } for (boolean doNext = true; this.isCommitLogAvailable() && doNext; ) { if (DefaultMessageStore.this.getMessageStoreConfig().isDuplicationEnable() && this.reputFromOffset >

= DefaultMessageStore.this.getConfirmOffset () {break;} / / 1. Get the data pointed to by the reputFromOffset offset SelectMappedBufferResult result = DefaultMessageStore.this.commitLog.getData (reputFromOffset); if (result! = null) {try {this.reputFromOffset = result.getStartOffset (); for (int readSize = 0; readSize)

< result.getSize() && doNext; ) { // 2. 解析消息体 DispatchRequest dispatchRequest = DefaultMessageStore.this.commitLog.checkMessageAndReturnSize(result.getByteBuffer(), false, false); int size = dispatchRequest.getBufferSize() == -1 ? dispatchRequest.getMsgSize() : dispatchRequest.getBufferSize(); if (dispatchRequest.isSuccess()) { if (size >

0) {/ / 3. Execute Distribution DefaultMessageStore.this.doDispatch (dispatchRequest); / /... This.reputFromOffset + = size; readSize + = size; / /...} else if (size = = 0) {this.reputFromOffset = DefaultMessageStore.this.commitLog.rollNextFile (this.reputFromOffset) ReadSize = result.getSize () }} else if (! dispatchRequest.isSuccess ()) {/ /...} finally {result.release () }} else {doNext = false;}

The insert operation of ConsumeQueue is as follows:

/ / ConsumeQueue.java private boolean putMessagePositionInfo (final long offset, final int size, final long tagsCode, final long cqOffset) {if (offset + size 0) {String [] keyset = keys.split (MessageConst.KEY_SEPARATOR); for (int I = 0; I

< keyset.length; i++) { // 为每个key执行写入 String key = keyset[i]; if (key.length() >

0) {indexFile = putKey (indexFile, msg, buildKey (topic, key)); if (indexFile = = null) {log.error ("putKey error commitlog {} uniqkey {}", req.getCommitLogOffset (), req.getUniqKey ()); return } else {log.error ("build index error, stop building index");}} private IndexFile putKey (IndexFile indexFile, DispatchRequest msg, String idxKey) {for (boolean ok = indexFile.putKey (idxKey, msg.getCommitLogOffset (), msg.getStoreTimestamp ());! ok ) {log.warn ("Index file [" + indexFile.getFileName () + "] is full, trying to create another one"); indexFile = retryGetAndCreateIndexFile (); / / File is full, retry if (null = = indexFile) {return null;} ok = indexFile.putKey (idxKey, msg.getCommitLogOffset (), msg.getStoreTimestamp ()) } return indexFile;}

The overall process of message dump is shown below:

After reading this article, I believe you have a certain understanding of "how to realize message dump in broker message storage in RocketMQ". If you want to know more about it, you are welcome to follow the industry information channel. Thank you for reading!

Welcome to subscribe "Shulou Technology Information " to get latest news, interesting things and hot topics in the IT industry, and controls the hottest and latest Internet news, technology news and IT industry trends.