• RocketMQ源码阅读（七）ConsumeQueue和IndexFile

    目录

    ConsumeQueue

    IndexFile

    putKey

    selectPhyOffset

    ConsumeQueue

    生产者将消息发给broker，broker将所有消息全部写入commitLog一个文件中，那消费者如何根据topic获取消息，RocketMQ使用的是逻辑上的topic队列，即ConsumeQueue。

    消息写入commitLog同时，将消息在commitLog的offset偏移量、消息大小、tags信息写入对应的topic队列（ConsumeQueue），消费者获取消息时先从ConsumeQueue获取消息的offset，在根据offset从commitLog中查询真正的消息数据。

    引入别人一张图直观了解：

    文件的存储结构：

    文件名是Topic名 

     

     一个Topic对应多个队列，名是queueId，每个文件夹都是一个ConsumeQueue

    一个队列中每个文件大小默认是6000000字节 

    实例化代码：

    根据topic文件路径和queueId创建一个MappedFileQueue

     核心方法：putMessagePositionInfoWrapper

    将请求的消息的offset,size,tags等写入MappedFileQueue

    1.    public void putMessagePositionInfoWrapper(DispatchRequest request) {
    2.         final int maxRetries = 30;
    3.         boolean canWrite = this.defaultMessageStore.getRunningFlags().isCQWriteable();
    4.         for (int i = 0; i < maxRetries && canWrite; i++) {
    5.             long tagsCode = request.getTagsCode();
    6.  
    7.             // 忽略部分无关代码
    8.  
    9.             boolean result = this.putMessagePositionInfo(request.getCommitLogOffset(),
    10.                 request.getMsgSize(), tagsCode, request.getConsumeQueueOffset());
    11.             if (result) {
    12.                 if (this.defaultMessageStore.getMessageStoreConfig().getBrokerRole() == BrokerRole.SLAVE ||
    13.                     this.defaultMessageStore.getMessageStoreConfig().isEnableDLegerCommitLog()) {
    14.                     this.defaultMessageStore.getStoreCheckpoint().setPhysicMsgTimestamp(request.getStoreTimestamp());
    15.                 }
    16.                 this.defaultMessageStore.getStoreCheckpoint().setLogicsMsgTimestamp(request.getStoreTimestamp());
    17.                 return;
    18.             } else {
    19.                 // XXX: warn and notify me
    20.                 log.warn("[BUG]put commit log position info to " + topic + ":" + queueId + " " + request.getCommitLogOffset()
    21.                     + " failed, retry " + i + " times");
    22.  
    23.                 try {
    24.                     Thread.sleep(1000);
    25.                 } catch (InterruptedException e) {
    26.                     log.warn("", e);
    27.                 }
    28.             }
    29.         }
    30.  
    31.         // XXX: warn and notify me
    32.         log.error("[BUG]consume queue can not write, {} {}", this.topic, this.queueId);
    33.         this.defaultMessageStore.getRunningFlags().makeLogicsQueueError();
    34.     }
    35.  
    36.  
    37.   private boolean putMessagePositionInfo(final long offset, final int size, final long tagsCode,
    38.         final long cqOffset) {
    39.  
    40.         if (offset + size <= this.maxPhysicOffset) {
    41.             log.warn("Maybe try to build consume queue repeatedly maxPhysicOffset={} phyOffset={}", maxPhysicOffset, offset);
    42.             return true;
    43.         }
    44.  
    45.         // 一条消息在consumeQueue中的大小为定长20字节
    46.         this.byteBufferIndex.flip();
    47.         this.byteBufferIndex.limit(CQ_STORE_UNIT_SIZE);
    48.         this.byteBufferIndex.putLong(offset);// 在commitLog中的偏移量  8字节
    49.         this.byteBufferIndex.putInt(size);// 消息大小  4字节
    50.         this.byteBufferIndex.putLong(tagsCode);// tags 标签的hashCode 8字节
    51.  
    52.         final long expectLogicOffset = cqOffset * CQ_STORE_UNIT_SIZE;
    53.  
    54.         MappedFile mappedFile = this.mappedFileQueue.getLastMappedFile(expectLogicOffset);
    55.         if (mappedFile != null) {
    56.  
    57.             if (mappedFile.isFirstCreateInQueue() && cqOffset != 0 && mappedFile.getWrotePosition() == 0) {
    58.                 this.minLogicOffset = expectLogicOffset;
    59.                 this.mappedFileQueue.setFlushedWhere(expectLogicOffset);
    60.                 this.mappedFileQueue.setCommittedWhere(expectLogicOffset);
    61.                 this.fillPreBlank(mappedFile, expectLogicOffset);
    62.                 log.info("fill pre blank space " + mappedFile.getFileName() + " " + expectLogicOffset + " "
    63.                     + mappedFile.getWrotePosition());
    64.             }
    65.  
    66.             if (cqOffset != 0) {
    67.                 long currentLogicOffset = mappedFile.getWrotePosition() + mappedFile.getFileFromOffset();
    68.  
    69.                 if (expectLogicOffset < currentLogicOffset) {
    70.                     log.warn("Build  consume queue repeatedly, expectLogicOffset: {} currentLogicOffset: {} Topic: {} QID: {} Diff: {}",
    71.                         expectLogicOffset, currentLogicOffset, this.topic, this.queueId, expectLogicOffset - currentLogicOffset);
    72.                     return true;
    73.                 }
    74.  
    75.             }
    76.             this.maxPhysicOffset = offset + size;
    77.             // 消息写入
    78.             return mappedFile.appendMessage(this.byteBufferIndex.array());
    79.         }
    80.         return false;
    81.     }

    IndexFile

    RocketMQ支持根据消息的key进行消息查询，和ConsumeQueue类似，如果消息中有key，就为这条消息建立Hash索引，即将消息在CommitLog中的偏移量写入索引文件，

     每个broker有一组indexFile，文件名即当时创建时的时间戳

    每个indexFile由三部分组成

    1、indexHeader, 长度为40字节，包含六个内容：biginTimestamp(第一条消息存储时间戳)，endTimestamp（最后一条消息存储时间戳），biginPhyoffset(第一条消息在commitlog中的偏移量，即commitlog offset)，endPhyoffset（最后一条消息在commitlog中的偏移量），hashSlotCount（含有index的slot数量），indexCount(包含的索引单元的个数) 

    2、slots,  参数控制，默认是500万，每个slot大小为4字节，槽中的4个字节存放的是索引块的位置

    3、index, 参数控制，默认是2000万，每个索引块大小为20字节， 4字节的hashKey，8字节的commitLog偏移量， 4字节的 当前key对应消息的存储时间与indexFile的时间差，4字节，当前索引块的前一个索引块位置

    1. public class IndexFile {
    2.     private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.STORE_LOGGER_NAME);
    3.     private static int hashSlotSize = 4;// 每个槽大小  4字节
    4.     private static int indexSize = 20;// 每个索引块大小， 20字节
    5.     private static int invalidIndex = 0;
    6.     private final int hashSlotNum; // 槽数量，默认500万
    7.     private final int indexNum;// 索引块数量，默认2000万
    8.     private final MappedFile mappedFile;
    9.     private final FileChannel fileChannel;
    10.     private final MappedByteBuffer mappedByteBuffer;
    11.     private final IndexHeader indexHeader;
    12.  
    13.     public IndexFile(final String fileName, final int hashSlotNum, final int indexNum,
    14.         final long endPhyOffset, final long endTimestamp) throws IOException {
    15.         // 文件总大小：header+500万的槽+2000万的索引块
    16.         int fileTotalSize =
    17.             IndexHeader.INDEX_HEADER_SIZE + (hashSlotNum * hashSlotSize) + (indexNum * indexSize);
    18.         this.mappedFile = new MappedFile(fileName, fileTotalSize);
    19.         this.fileChannel = this.mappedFile.getFileChannel();
    20.         this.mappedByteBuffer = this.mappedFile.getMappedByteBuffer();
    21.         this.hashSlotNum = hashSlotNum;
    22.         this.indexNum = indexNum;
    23.  
    24.         ByteBuffer byteBuffer = this.mappedByteBuffer.slice();
    25.  
    26.         // 取前40个字节封装为IndexHeader
    27.         this.indexHeader = new IndexHeader(byteBuffer);
    28.  
    29.         if (endPhyOffset > 0) {
    30.             this.indexHeader.setBeginPhyOffset(endPhyOffset);
    31.             this.indexHeader.setEndPhyOffset(endPhyOffset);
    32.         }
    33.  
    34.         if (endTimestamp > 0) {
    35.             this.indexHeader.setBeginTimestamp(endTimestamp);
    36.             this.indexHeader.setEndTimestamp(endTimestamp);
    37.         }
    38.     }
    39. }

    putKey

    将某个消息key和对应commitLog的offset写入indexFile 

    1.     public boolean putKey(final String key, final long phyOffset, final long storeTimestamp) {
    2.         // 头中记录的索引块数量 小于 文件最大值
    3.         if (this.indexHeader.getIndexCount() < this.indexNum) {
    4.             // 对key的hashCode取模，算出slot的位置
    5.             int keyHash = indexKeyHashMethod(key);
    6.             int slotPos = keyHash % this.hashSlotNum;
    7.             int absSlotPos = IndexHeader.INDEX_HEADER_SIZE + slotPos * hashSlotSize;
    8.  
    9.             FileLock fileLock = null;
    10.  
    11.             try {
    12.  
    13.                 // fileLock = this.fileChannel.lock(absSlotPos, hashSlotSize,
    14.                 // false);
    15.                 // 取对应位置槽的值，也就是槽对应的索引块位置
    16.                 int slotValue = this.mappedByteBuffer.getInt(absSlotPos);
    17.                 if (slotValue <= invalidIndex || slotValue > this.indexHeader.getIndexCount()) {
    18.                     slotValue = invalidIndex;
    19.                 }
    20.  
    21.                 long timeDiff = storeTimestamp - this.indexHeader.getBeginTimestamp();
    22.  
    23.                 timeDiff = timeDiff / 1000;
    24.  
    25.                 if (this.indexHeader.getBeginTimestamp() <= 0) {
    26.                     timeDiff = 0;
    27.                 } else if (timeDiff > Integer.MAX_VALUE) {
    28.                     timeDiff = Integer.MAX_VALUE;
    29.                 } else if (timeDiff < 0) {
    30.                     timeDiff = 0;
    31.                 }
    32.  
    33.                 // 当前索引块末尾空闲的位置
    34.                 int absIndexPos =
    35.                     IndexHeader.INDEX_HEADER_SIZE + this.hashSlotNum * hashSlotSize
    36.                         + this.indexHeader.getIndexCount() * indexSize;
    37.  
    38.                 // 写入索引块的数据
    39.                 this.mappedByteBuffer.putInt(absIndexPos, keyHash);// 4字节的hashKey
    40.                 this.mappedByteBuffer.putLong(absIndexPos + 4, phyOffset);// 8字节的commitLog偏移量
    41.                 this.mappedByteBuffer.putInt(absIndexPos + 4 + 8, (int) timeDiff);// 4字节的 当前key对应消息的存储时间与indexFile的时间差
    42.                 this.mappedByteBuffer.putInt(absIndexPos + 4 + 8 + 4, slotValue);// 4字节，当前索引块的前一个索引块位置
    43.  
    44.                 // 更新槽上的索引块位置
    45.                 this.mappedByteBuffer.putInt(absSlotPos, this.indexHeader.getIndexCount());
    46.  
    47.                 // 更新indexHeader中的各种标志位
    48.                 if (this.indexHeader.getIndexCount() <= 1) {
    49.                     this.indexHeader.setBeginPhyOffset(phyOffset);
    50.                     this.indexHeader.setBeginTimestamp(storeTimestamp);
    51.                 }
    52.  
    53.                 if (invalidIndex == slotValue) {
    54.                     this.indexHeader.incHashSlotCount();
    55.                 }
    56.                 this.indexHeader.incIndexCount();
    57.                 this.indexHeader.setEndPhyOffset(phyOffset);
    58.                 this.indexHeader.setEndTimestamp(storeTimestamp);
    59.  
    60.                 return true;
    61.             } catch (Exception e) {
    62.                 log.error("putKey exception, Key: " + key + " KeyHashCode: " + key.hashCode(), e);
    63.             } finally {
    64.                 if (fileLock != null) {
    65.                     try {
    66.                         fileLock.release();
    67.                     } catch (IOException e) {
    68.                         log.error("Failed to release the lock", e);
    69.                     }
    70.                 }
    71.             }
    72.         } else {
    73.             log.warn("Over index file capacity: index count = " + this.indexHeader.getIndexCount()
    74.                 + "; index max num = " + this.indexNum);
    75.         }
    76.  
    77.         return false;
    78.     }

    selectPhyOffset

    从indexFile根据某个key查询符合条件的所有消息，返回List<Long> ，列表中的值是消息在commitLog中的偏移量

    1.    public void selectPhyOffset(final List<Long> phyOffsets, final String key, final int maxNum,
    2.         final long begin, final long end, boolean lock) {
    3.         if (this.mappedFile.hold()) {
    4.             // 获取hashKey所对应的槽位置
    5.             int keyHash = indexKeyHashMethod(key);
    6.             int slotPos = keyHash % this.hashSlotNum;
    7.             int absSlotPos = IndexHeader.INDEX_HEADER_SIZE + slotPos * hashSlotSize;
    8.  
    9.             FileLock fileLock = null;
    10.             try {
    11.                 if (lock) {
    12.                     // fileLock = this.fileChannel.lock(absSlotPos,
    13.                     // hashSlotSize, true);
    14.                 }
    15.  
    16.                 int slotValue = this.mappedByteBuffer.getInt(absSlotPos);
    17.                 // if (fileLock != null) {
    18.                 // fileLock.release();
    19.                 // fileLock = null;
    20.                 // }
    21.  
    22.                 if (slotValue <= invalidIndex || slotValue > this.indexHeader.getIndexCount()
    23.                     || this.indexHeader.getIndexCount() <= 1) {
    24.                 } else {
    25.                     for (int nextIndexToRead = slotValue; ; ) {
    26.                         if (phyOffsets.size() >= maxNum) {
    27.                             break;
    28.                         }
    29.  
    30.                         // 获取槽对应的index块位置，索引块是根据后4字节被串成了一个链表，需要下边循环查询
    31.                         int absIndexPos =
    32.                             IndexHeader.INDEX_HEADER_SIZE + this.hashSlotNum * hashSlotSize
    33.                                 + nextIndexToRead * indexSize;
    34.  
    35.                         int keyHashRead = this.mappedByteBuffer.getInt(absIndexPos);
    36.                         long phyOffsetRead = this.mappedByteBuffer.getLong(absIndexPos + 4);
    37.  
    38.                         long timeDiff = (long) this.mappedByteBuffer.getInt(absIndexPos + 4 + 8);
    39.                         int prevIndexRead = this.mappedByteBuffer.getInt(absIndexPos + 4 + 8 + 4);// 前一个索引块
    40.  
    41.                         if (timeDiff < 0) {
    42.                             break;
    43.                         }
    44.  
    45.                         timeDiff *= 1000L;
    46.  
    47.                         // commitLog中消息的存储时间需要在参数begin和end之间
    48.                         long timeRead = this.indexHeader.getBeginTimestamp() + timeDiff;
    49.                         boolean timeMatched = (timeRead >= begin) && (timeRead <= end);
    50.  
    51.                         // 如果keyHash相同，即选中加入list
    52.                         if (keyHash == keyHashRead && timeMatched) {
    53.                             phyOffsets.add(phyOffsetRead);
    54.                         }
    55.  
    56.                         // 在循环前一个索引块
    57.                         if (prevIndexRead <= invalidIndex
    58.                             || prevIndexRead > this.indexHeader.getIndexCount()
    59.                             || prevIndexRead == nextIndexToRead || timeRead < begin) {
    60.                             break;
    61.                         }
    62.  
    63.                         nextIndexToRead = prevIndexRead;
    64.                     }
    65.                 }
    66.             } catch (Exception e) {
    67.                 log.error("selectPhyOffset exception ", e);
    68.             } finally {
    69.                 if (fileLock != null) {
    70.                     try {
    71.                         fileLock.release();
    72.                     } catch (IOException e) {
    73.                         log.error("Failed to release the lock", e);
    74.                     }
    75.                 }
    76.  
    77.                 this.mappedFile.release();
    78.             }
    79.         }
    80.     }

  • 相关阅读:
    MyBatis与Spring框架整合实现对数据库的增删改查
    基于python-socket的端口扫描
    修改例程flags_asyncio.py使能在python311环境下运行
    rpc的正确打开方式｜读懂Go原生net/rpc包
    Docker 01 概述
    短视频账号矩阵系统saas管理私信回复管理系统
    吉时利2602A数字源表-安泰测试
    独处是一种修行
    湖北大学2024年成人高考函授报名高起专大数据与会计专业介绍
    uniapp-历史搜索记录
  • 原文地址：https://blog.csdn.net/xyjy11/article/details/125407306