通過(guò)這三個(gè)文件徹底搞懂rocketmq的存儲(chǔ)原理

  @Override    public PutMessageResult putMessage(MessageExtBrokerInner msg) {        if (this.shutdown) {            log.warn("message store has shutdown, so putMessage is forbidden");            return new PutMessageResult(PutMessageStatus.SERVICE_NOT_AVAILABLE, null);        }
        // 從節(jié)點(diǎn)不允許寫入        if (BrokerRole.SLAVE == this.messageStoreConfig.getBrokerRole()) {            long value = this.printTimes.getAndIncrement();            if ((value % 50000) == 0) {                log.warn("message store is slave mode, so putMessage is forbidden ");            }
            return new PutMessageResult(PutMessageStatus.SERVICE_NOT_AVAILABLE, null);        }
        // store是否允許寫入        if (!this.runningFlags.isWriteable()) {            long value = this.printTimes.getAndIncrement();            if ((value % 50000) == 0) {                log.warn("message store is not writeable, so putMessage is forbidden " + this.runningFlags.getFlagBits());            }
            return new PutMessageResult(PutMessageStatus.SERVICE_NOT_AVAILABLE, null);        } else {            this.printTimes.set(0);        }
        // topic過(guò)長(zhǎng)        if (msg.getTopic().length() > Byte.MAX_VALUE) {            log.warn("putMessage message topic length too long " + msg.getTopic().length());            return new PutMessageResult(PutMessageStatus.MESSAGE_ILLEGAL, null);        }
        // 消息附加屬性過(guò)長(zhǎng)        if (msg.getPropertiesString() != null && msg.getPropertiesString().length() > Short.MAX_VALUE) {            log.warn("putMessage message properties length too long " + msg.getPropertiesString().length());            return new PutMessageResult(PutMessageStatus.PROPERTIES_SIZE_EXCEEDED, null);        }
        if (this.isOSPageCacheBusy()) {            return new PutMessageResult(PutMessageStatus.OS_PAGECACHE_BUSY, null);        }
        long beginTime = this.getSystemClock().now();        // 添加消息到commitLog        PutMessageResult result = this.commitLog.putMessage(msg);
        long eclipseTime = this.getSystemClock().now() - beginTime;        if (eclipseTime > 500) {            log.warn("putMessage not in lock eclipse time(ms)={}, bodyLength={}", eclipseTime, msg.getBody().length);        }        this.storeStatsService.setPutMessageEntireTimeMax(eclipseTime);
        if (null == result || !result.isOk()) {            this.storeStatsService.getPutMessageFailedTimes().incrementAndGet();        }
        return result;    }

public void run() {            while (!this.isStopped()) {                try {                    Thread.sleep(1);                    this.doReput(); // 構(gòu)建ComsumerQueue                } catch (Exception e) {                    DefaultMessageStore.log.warn(this.getServiceName() + " service has exception. ", e);                }            }        }

private void doReput() {            for (boolean doNext = true; this.isCommitLogAvailable() && doNext; ) {                SelectMappedBufferResult result = DefaultMessageStore.this.commitLog.getData(reputFromOffset);// 拿到所有的最新寫入CommitLog的數(shù)據(jù)                if (result != null) {                    try {                        this.reputFromOffset = result.getStartOffset();
                        for (int readSize = 0; readSize < result.getSize() && doNext; ) {                            DispatchRequest dispatchRequest =                            DefaultMessageStore.this.commitLog.checkMessageAndReturnSize(result.getByteBuffer(), false, false); // 一條一條的讀消息                            int size = dispatchRequest.getMsgSize();
                            if (dispatchRequest.isSuccess()) {                                if (size > 0) {                                    DefaultMessageStore.this.doDispatch(dispatchRequest); // 派發(fā)消息，進(jìn)行處理，其中就包括構(gòu)建ComsumerQueue                                    this.reputFromOffset += size;                                    readSize += size;                                } else if (size == 0) { //                                     this.reputFromOffset = DefaultMessageStore.this.commitLog.rollNextFile(this.reputFromOffset);                                    readSize = result.getSize();                                }                            } else if (!dispatchRequest.isSuccess()) { // 獲取消息異常
                                if (size > 0) {                                    log.error("[BUG]read total count not equals msg total size. reputFromOffset={}", reputFromOffset);                                    this.reputFromOffset += size;                                } else {                                    doNext = false;                                    if (DefaultMessageStore.this.brokerConfig.getBrokerId() == MixAll.MASTER_ID) {                                        this.reputFromOffset += result.getSize() - readSize;                                    }                                }                            }                        }                    } finally {                        result.release();                    }                } else {                    doNext = false;                }            }        }

//8位 該索引文件的第一個(gè)消息(Message)的存儲(chǔ)時(shí)間(落盤時(shí)間)this.byteBuffer.putLong(beginTimestampIndex, this.beginTimestamp.get());//8位 該索引文件的最后一個(gè)消息(Message)的存儲(chǔ)時(shí)間(落盤時(shí)間)this.byteBuffer.putLong(endTimestampIndex, this.endTimestamp.get());//8位 該索引文件第一個(gè)消息(Message)的在CommitLog(消息存儲(chǔ)文件)的物理位置偏移量(可以通過(guò)該物理偏移直接獲取到該消息)this.byteBuffer.putLong(beginPhyoffsetIndex, this.beginPhyOffset.get());//8位 該索引文件最后一個(gè)消息(Message)的在CommitLog(消息存儲(chǔ)文件)的物理位置偏移量this.byteBuffer.putLong(endPhyoffsetIndex, this.endPhyOffset.get());//4位 該索引文件目前的hash slot的個(gè)數(shù)this.byteBuffer.putInt(hashSlotcountIndex, this.hashSlotCount.get());//4位 索引文件目前的索引個(gè)數(shù)this.byteBuffer.putInt(indexCountIndex, this.indexCount.get());

//slot的數(shù)據(jù)存放位置 40 + keyHash %（500W）* 4int absSlotPos = IndexHeader.INDEX_HEADER_SIZE + slotPos * hashSlotSize;
//Slot Table//4字節(jié)//記錄該slot當(dāng)前index，如果hash沖突（即absSlotPos一致）作為下一次該slot新增的前置indexthis.mappedByteBuffer.putInt(absSlotPos, this.indexHeader.getIndexCount());

//Index Linked list//topic+message key的hash值this.mappedByteBuffer.putInt(absIndexPos, keyHash);//消息在CommitLog的物理文件地址, 可以直接查詢到該消息(索引的核心機(jī)制)this.mappedByteBuffer.putLong(absIndexPos + 4, phyOffset);//消息的落盤時(shí)間與header里的beginTimestamp的差值(為了節(jié)省存儲(chǔ)空間，如果直接存message的落盤時(shí)間就得8bytes)this.mappedByteBuffer.putInt(absIndexPos + 4 + 8, (int) timeDiff);//9、記錄該slot上一個(gè)index//hash沖突處理的關(guān)鍵之處, 相同hash值上一個(gè)消息索引的index(如果當(dāng)前消息索引是該hash值的第一個(gè)索引，則prevIndex=0, 也是消息索引查找時(shí)的停止條件)，每個(gè)slot位置的第一個(gè)消息的prevIndex就是0的this.mappedByteBuffer.putInt(absIndexPos + 4 + 8 + 4, slotValue);

public boolean putKey(final String key, final long phyOffset, final long storeTimestamp) {        //1. 判斷該索引文件的索引數(shù)小于最大的索引數(shù)，如果>=最大索引數(shù)，IndexService就會(huì)嘗試新建一個(gè)索引文件        if (this.indexHeader.getIndexCount() < this.indexNum) {            //2. 計(jì)算該message key的hash值            int keyHash = indexKeyHashMethod(key);            //3. 根據(jù)message key的hash值散列到某個(gè)hash slot里            int slotPos = keyHash % this.hashSlotNum;            //4. 計(jì)算得到該hash slot的實(shí)際文件位置Position            int absSlotPos = IndexHeader.INDEX_HEADER_SIZE + slotPos * hashSlotSize;
            try {                //5. 根據(jù)該hash slot的實(shí)際文件位置absSlotPos得到slot里的值                //這里有兩種情況：                //1). slot=0, 當(dāng)前message的key是該hash值第一個(gè)消息索引                //2). slot>0, 該key hash值上一個(gè)消息索引的位置                int slotValue = this.mappedByteBuffer.getInt(absSlotPos);
                //6. 數(shù)據(jù)校驗(yàn)及修正                if (slotValue <= invalidIndex || slotValue > this.indexHeader.getIndexCount()) {                    slotValue = invalidIndex;                }
                long timeDiff = storeTimestamp - this.indexHeader.getBeginTimestamp();
                timeDiff = timeDiff / 1000;
                if (this.indexHeader.getBeginTimestamp() <= 0) {                    timeDiff = 0;                } else if (timeDiff > Integer.MAX_VALUE) {                    timeDiff = Integer.MAX_VALUE;                } else if (timeDiff < 0) {                    timeDiff = 0;                }
                //7. 計(jì)算當(dāng)前消息索引具體的存儲(chǔ)位置(Append模式)                int absIndexPos =                    IndexHeader.INDEX_HEADER_SIZE + this.hashSlotNum * hashSlotSize                        + this.indexHeader.getIndexCount() * indexSize;                //8. 存入該消息索引                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);
                //9. 關(guān)鍵之處：在該key hash slot處存入當(dāng)前消息索引的位置，下次通過(guò)該key進(jìn)行搜索時(shí)                //會(huì)找到該key hash slot -> slot value -> curIndex ->                 //if(curIndex.prevIndex>0) pre index (一直循環(huán) 直至該curIndex.prevIndex==0就停止)                this.mappedByteBuffer.putInt(absSlotPos, this.indexHeader.getIndexCount());
                if (this.indexHeader.getIndexCount() <= 1) {                    this.indexHeader.setBeginPhyOffset(phyOffset);                    this.indexHeader.setBeginTimestamp(storeTimestamp);                }
                this.indexHeader.incHashSlotCount();                this.indexHeader.incIndexCount();                this.indexHeader.setEndPhyOffset(phyOffset);                this.indexHeader.setEndTimestamp(storeTimestamp);
                return true;            } catch (Exception e) {                log.error("putKey exception, Key: " + key + " KeyHashCode: " + key.hashCode(), e);            }         } else {            log.warn("Over index file capacity: index count = " + this.indexHeader.getIndexCount()                + "; index max num = " + this.indexNum);        }
        return false; }

public void selectPhyOffset(final List phyOffsets, final String key, final int maxNum,        final long begin, final long end, boolean lock) {        if (this.mappedFile.hold()) {            //1. 計(jì)算該key的hash            int keyHash = indexKeyHashMethod(key);            //2. 計(jì)算該hash value 對(duì)應(yīng)的hash slot位置            int slotPos = keyHash % this.hashSlotNum;            //3. 計(jì)算該hash value 對(duì)應(yīng)的hash slot物理文件位置            int absSlotPos = IndexHeader.INDEX_HEADER_SIZE + slotPos * hashSlotSize;
            FileLock fileLock = null;            try {                //4. 取出該hash slot 的值                int slotValue = this.mappedByteBuffer.getInt(absSlotPos);
                //5. 該slot value <= 0 就代表沒(méi)有該key對(duì)應(yīng)的消息索引,直接結(jié)束搜索                //   該slot value > maxIndexCount 就代表該key對(duì)應(yīng)的消息索引超過(guò)最大限制，數(shù)據(jù)有誤,直接結(jié)束搜索                if (slotValue <= invalidIndex || slotValue > this.indexHeader.getIndexCount()                    || this.indexHeader.getIndexCount() <= 1) {                } else {                    //6. 從當(dāng)前slot value 開(kāi)始搜索                    for (int nextIndexToRead = slotValue; ; ) {                        if (phyOffsets.size() >= maxNum) {                            break;                        }
                        //7. 找到當(dāng)前slot value(也就是index count)物理文件位置                        int absIndexPos =                            IndexHeader.INDEX_HEADER_SIZE + this.hashSlotNum * hashSlotSize                                + nextIndexToRead * indexSize;
                        //8. 讀取消息索引數(shù)據(jù)                        int keyHashRead = this.mappedByteBuffer.getInt(absIndexPos);                        long phyOffsetRead = this.mappedByteBuffer.getLong(absIndexPos + 4);
                        long timeDiff = (long) this.mappedByteBuffer.getInt(absIndexPos + 4 + 8);                        //9. 獲取該消息索引的上一個(gè)消息索引index(可以看成鏈表的prev 指向上一個(gè)鏈節(jié)點(diǎn)的引用)                        int prevIndexRead = this.mappedByteBuffer.getInt(absIndexPos + 4 + 8 + 4);                        //10. 數(shù)據(jù)校驗(yàn)                        if (timeDiff < 0) {                            break;                        }
                        timeDiff *= 1000L;
                        long timeRead = this.indexHeader.getBeginTimestamp() + timeDiff;                        boolean timeMatched = (timeRead >= begin) && (timeRead <= end);                        //10. 數(shù)據(jù)校驗(yàn)比對(duì) hash值和落盤時(shí)間                        if (keyHash == keyHashRead && timeMatched) {                            phyOffsets.add(phyOffsetRead);                        }
                        //當(dāng)prevIndex <= 0 或prevIndex > maxIndexCount 或prevIndexRead == nextIndexToRead 或 timeRead < begin 停止搜索                        if (prevIndexRead <= invalidIndex                            || prevIndexRead > this.indexHeader.getIndexCount()                            || prevIndexRead == nextIndexToRead || timeRead < begin) {                            break;                        }
                        nextIndexToRead = prevIndexRead;                    }                }            } catch (Exception e) {                log.error("selectPhyOffset exception ", e);            } finally {
                this.mappedFile.release();            }        }    }