Index索引文件及构建

Index索引文件概览

消息消费队列是RocetMQ专门为消息订阅构建的索引服务，提高主题与消息队列检索消息的速度。IndexFile（索引文件）提供了一种可以通过key或时间区间来查询消息的方法。Index文件的存储位置是：$HOME/store/index/{fileName}，文件名fileName是以创建时的时间戳命名的，固定的单个IndexFile文件大小约为400M，一个IndexFile可以保存 2000W个索引，IndexFile的底层存储设计为在文件系统中实现HashMap结构，故rocketmq的索引文件其底层实现为hash索引。

Index索引文件结构：

image.png

IndexFile异步构造

构建consumequeue、indexFile索引文件，通过一个ReputMessageService异步线程进行处理，构建consumequeue、indexFile索引文件的数据从commitLog的MappedFile中的ByteBuffer中获取，一条消息消息构造一个构建索引服务的DispatchRequest请求，再由ConsumeQueue服务处理DispatchRequest请求构建consumequeue的mappedFile文件。由IndexService处理请求构建indexFile索引文件，然后将各自的文件进行刷盘。

IndexFile

IndexFile对象的主要成员属性，其中包含MappedFile对象，这个和ConsumeQueue中利用的MappedFile是一样的作用，用来做磁盘IO，将内存映射消息写入磁盘。

public class IndexFile {
    // hash槽的大小
    private static int hashSlotSize = 4;
    // 索引大小
    private static int indexSize = 20;
    private static int invalidIndex = 0;
    // 索引槽的数量 500万
    private final int hashSlotNum;
    // 索引的数量 2千万
    private final int indexNum;
    // 索引的MappedFile
    private final MappedFile mappedFile;
    private final FileChannel fileChannel;
    //MappedFile中的直接内存，用来存放hash索引用
    private final MappedByteBuffer mappedByteBuffer;
    // 索引头
    private final IndexHeader indexHeader;
} 

IndexHeader

每一个IndexFile都包含IndexHeader，是这个IndexFile的汇总信息。

// 存放IndexHeader的buffer
private final ByteBuffer byteBuffer;
// indexFile存放消息的存储时间的 开始时间
private AtomicLong beginTimestamp = new AtomicLong(0);
// indexFile存放消息的存储时间的 结束时间
private AtomicLong endTimestamp = new AtomicLong(0);
// indexFile 开始commitLog 文件的offset
private AtomicLong beginPhyOffset = new AtomicLong(0);
// IndexFile 结束commitLog 文件的offset
private AtomicLong endPhyOffset = new AtomicLong(0);
// hash槽的Count 500万
private AtomicInteger hashSlotCount = new AtomicInteger(0);
// hash索引的条目2千万条 500 * 40
private AtomicInteger indexCount = new AtomicInteger(1);

IndexService

IndexService 是构建Index索引文件的服务。一条消息可能有多个key，这样这条消息就会有多个索引条目。

根据一条消息的请求构建Index索引，

/**
 * 根据一条消息的请求构建Index索引，并放入
 * @param req
 */
public void buildIndex(DispatchRequest req) {
    //获取需要写入的IndexFile
    IndexFile indexFile = retryGetAndCreateIndexFile();
    if (indexFile != null) {
        // 文件结尾offset
        long endPhyOffset = indexFile.getEndPhyOffset();
        DispatchRequest msg = req;
        String topic = msg.getTopic();
        String keys = msg.getKeys();
        if (msg.getCommitLogOffset() < endPhyOffset) {
            return;
        }
        // 事务消息
        final int tranType = MessageSysFlag.getTransactionValue(msg.getSysFlag());
        switch (tranType) {
            case MessageSysFlag.TRANSACTION_NOT_TYPE:
            case MessageSysFlag.TRANSACTION_PREPARED_TYPE:
            case MessageSysFlag.TRANSACTION_COMMIT_TYPE:
                break;
            case MessageSysFlag.TRANSACTION_ROLLBACK_TYPE:
                return;
        }
        // 根据uniqKey来构建文件
        if (req.getUniqKey() != null) {
            indexFile = putKey(indexFile, msg, buildKey(topic, req.getUniqKey()));
            if (indexFile == null) {
                log.error("putKey error commitlog {} uniqkey {}", req.getCommitLogOffset(), req.getUniqKey());
                return;
            }
        }
        // 根据keys中的每个key来构建indexFile
        if (keys != null && keys.length() > 0) {
            String[] keyset = keys.split(MessageConst.KEY_SEPARATOR);
            for (int i = 0; i < keyset.length; i++) {
                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");
    }
}

/**
 * 放置key到indexFile中
 * @param indexFile
 * @param msg 消息请求
 * @param idxKey topic#uniqKey
 * @return
 */
private IndexFile putKey(IndexFile indexFile, DispatchRequest msg, String idxKey) {
    // for循环一直方法详细，直到出现一个错误
    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
        indexFile = retryGetAndCreateIndexFile();
        if (null == indexFile) {
            return null;
        }
        //存放消息
        ok = indexFile.putKey(idxKey, msg.getCommitLogOffset(), msg.getStoreTimestamp());
    }

    return indexFile;
}

IndexFile存放这个消息的索引方法，计算key的hash值，确定hash槽，将索引信息存入MappedByteBuffer等待刷盘操作。

/**
 * 存放索引信息到mappedByteBuffer等待刷盘
 * @param key 存放的key
 * @param phyOffset 存放的物理offset
 * @param storeTimestamp 存放消息存储的时间
 * @return
 */
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;
        // 40byte hashIndex + 槽的位置 * 槽的大小
        int absSlotPos = IndexHeader.INDEX_HEADER_SIZE + slotPos * hashSlotSize;

        FileLock fileLock = null;

        try {

            // fileLock = this.fileChannel.lock(absSlotPos, hashSlotSize,
            // false);
            // 获取mappedByteBuffer写入的位置,上一个hash索引存储的位置
            int slotValue = this.mappedByteBuffer.getInt(absSlotPos);
            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;
            }
            // 存储的绝对位置
            int absIndexPos =
                IndexHeader.INDEX_HEADER_SIZE + this.hashSlotNum * hashSlotSize
                    + this.indexHeader.getIndexCount() * indexSize;
            // 存放hashKey，absIndexPos这个位置放入keyHash
            this.mappedByteBuffer.putInt(absIndexPos, keyHash);
            // 存放物理commitlog消息的物理offset，absIndexPos + 4（上一个keyhash占用了4个位置）这个位置放入phyOffset
            this.mappedByteBuffer.putLong(absIndexPos + 4, phyOffset);
            // 存放时间差值
            this.mappedByteBuffer.putInt(absIndexPos + 4 + 8, (int) timeDiff);
            // 存放上一个索引的位置
            this.mappedByteBuffer.putInt(absIndexPos + 4 + 8 + 4, slotValue);
            // 更新存放的索引数量，absSlotPos为索引槽的位置
            this.mappedByteBuffer.putInt(absSlotPos, this.indexHeader.getIndexCount());
            //更新文件索引的头信息，hash槽的总数、index条目的总数、最后消息的物理偏移量、最后消息的存储时间
            if (this.indexHeader.getIndexCount() <= 1) {
                this.indexHeader.setBeginPhyOffset(phyOffset);
                this.indexHeader.setBeginTimestamp(storeTimestamp);
            }

            if (invalidIndex == slotValue) {
                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);
        } finally {
            if (fileLock != null) {
                try {
                    fileLock.release();
                } catch (IOException e) {
                    log.error("Failed to release the lock", e);
                }
            }
        }
    } else {
        log.warn("Over index file capacity: index count = " + this.indexHeader.getIndexCount()
            + "; index max num = " + this.indexNum);
    }

    return false;
}

消息查询

QueryMessageProcessor接受客户端查询请求，进行处理。然后到DefaultMessageStore#queryMessage()方法，然后到IndexService#queryOffset(),最后到IndexFile#selectPhyOffset()方法。

// 根据 topic、key 查找到 indexFile 索引文件中的一条记录，根据其中的commitLog offset从CommitLog文件中读取消息的实体内容。
final QueryMessageResult queryMessageResult =
    this.brokerController.getMessageStore().queryMessage(requestHeader.getTopic(),
        requestHeader.getKey(), requestHeader.getMaxNum(), requestHeader.getBeginTimestamp(),
        requestHeader.getEndTimestamp());

// 根据 key 查询消息
@Override
public QueryMessageResult queryMessage(String topic, String key, int maxNum, long begin, long end) {
    QueryMessageResult queryMessageResult = new QueryMessageResult();

    long lastQueryMsgTime = end;

    for (int i = 0; i < 3; i++) {
        QueryOffsetResult queryOffsetResult = this.indexService.queryOffset(topic, key, maxNum, begin, lastQueryMsgTime);
        if (queryOffsetResult.getPhyOffsets().isEmpty()) {
            break;
        }

        Collections.sort(queryOffsetResult.getPhyOffsets());

        queryMessageResult.setIndexLastUpdatePhyoffset(queryOffsetResult.getIndexLastUpdatePhyoffset());
        queryMessageResult.setIndexLastUpdateTimestamp(queryOffsetResult.getIndexLastUpdateTimestamp());

        for (int m = 0; m < queryOffsetResult.getPhyOffsets().size(); m++) {
            long offset = queryOffsetResult.getPhyOffsets().get(m);

            try {

                boolean match = true;
                MessageExt msg = this.lookMessageByOffset(offset);
                if (0 == m) {
                    lastQueryMsgTime = msg.getStoreTimestamp();
                }

//                    String[] keyArray = msg.getKeys().split(MessageConst.KEY_SEPARATOR);
//                    if (topic.equals(msg.getTopic())) {
//                        for (String k : keyArray) {
//                            if (k.equals(key)) {
//                                match = true;
//                                break;
//                            }
//                        }
//                    }

                if (match) {
                    SelectMappedBufferResult result = this.commitLog.getData(offset, false);
                    if (result != null) {
                        int size = result.getByteBuffer().getInt(0);
                        result.getByteBuffer().limit(size);
                        result.setSize(size);
                        queryMessageResult.addMessage(result);
                    }
                } else {
                    log.warn("queryMessage hash duplicate, {} {}", topic, key);
                }
            } catch (Exception e) {
                log.error("queryMessage exception", e);
            }
        }

        if (queryMessageResult.getBufferTotalSize() > 0) {
            break;
        }

        if (lastQueryMsgTime < begin) {
            break;
        }
    }

    return queryMessageResult;
}

/**
 * 根据Index消息查询服务
 * @param topic
 * @param key
 * @param maxNum
 * @param begin 消息存储的开始时间，和IndexFile 的IndexHeader中的存储消息的开始时间进行对比
 * @param end 消息存储的结束时间，和IndexFile 的IndexHeader中的存储消息的结束时间进行对比
 * @return
 */
public QueryOffsetResult queryOffset(String topic, String key, int maxNum, long begin, long end) {
    List<Long> phyOffsets = new ArrayList<Long>(maxNum);

    long indexLastUpdateTimestamp = 0;
    long indexLastUpdatePhyoffset = 0;
    maxNum = Math.min(maxNum, this.defaultMessageStore.getMessageStoreConfig().getMaxMsgsNumBatch());
    try {
        this.readWriteLock.readLock().lock();
        if (!this.indexFileList.isEmpty()) {
            for (int i = this.indexFileList.size(); i > 0; i--) {
                IndexFile f = this.indexFileList.get(i - 1);
                boolean lastFile = i == this.indexFileList.size();
                if (lastFile) {
                    indexLastUpdateTimestamp = f.getEndTimestamp();
                    indexLastUpdatePhyoffset = f.getEndPhyOffset();
                }

                if (f.isTimeMatched(begin, end)) {

                    f.selectPhyOffset(phyOffsets, buildKey(topic, key), maxNum, begin, end, lastFile);
                }

                if (f.getBeginTimestamp() < begin) {
                    break;
                }

                if (phyOffsets.size() >= maxNum) {
                    break;
                }
            }
        }
    } catch (Exception e) {
        log.error("queryMsg exception", e);
    } finally {
        this.readWriteLock.readLock().unlock();
    }

    return new QueryOffsetResult(phyOffsets, indexLastUpdateTimestamp, indexLastUpdatePhyoffset);
}

/**
 * 根据索引key查找消息
 * @param phyOffsets 查找到的消息物理偏移量
 * @param key 索引key
 * @param maxNum 本次查找最大消息条数
 * @param begin 开始时间戳
 * @param end 结束时间戳
 * @param lock
 */
public void selectPhyOffset(final List<Long> phyOffsets, final String key, final int maxNum,
    final long begin, final long end, boolean lock) {
    if (this.mappedFile.hold()) {
        // 计算key的hashcode
        int keyHash = indexKeyHashMethod(key);
        // 定位到hash槽的位置
        int slotPos = keyHash % this.hashSlotNum;
        // 计算hash槽的绝对物理位置
        int absSlotPos = IndexHeader.INDEX_HEADER_SIZE + slotPos * hashSlotSize;

        FileLock fileLock = null;
        try {
            if (lock) {
                // fileLock = this.fileChannel.lock(absSlotPos,
                // hashSlotSize, true);
            }
            // 得到槽的值
            int slotValue = this.mappedByteBuffer.getInt(absSlotPos);
            // if (fileLock != null) {
            // fileLock.release();
            // fileLock = null;
            // }
            // 如果对应的Hash槽中存储的数据小于1或者大于当前索引条目个数则表示给HashCode没有对应的条条目，直接返回
            if (slotValue <= invalidIndex || slotValue > this.indexHeader.getIndexCount()
                || this.indexHeader.getIndexCount() <= 1) {
            } else {
                // 由于会存在hash冲突，根据slotValue定位该hash槽最新的一个Item条目，将存储的物理偏移加入到phyOffsets中，
                // 然后继续验证Item条目中存储的上一个Index下标，如果大于等于1并且小于最大条目数，则继续查找，否则结束查找。
                for (int nextIndexToRead = slotValue; ; ) {
                    // 如果大于查找数量，中断
                    if (phyOffsets.size() >= maxNum) {
                        break;
                    }
                    // 根据Index下标定位到条目的起始物理偏移量，然后依次读取hashCode、物理偏移量、时间差、上一个条目的Index下标
                    int absIndexPos =
                        IndexHeader.INDEX_HEADER_SIZE + this.hashSlotNum * hashSlotSize
                            + nextIndexToRead * indexSize;

                    int keyHashRead = this.mappedByteBuffer.getInt(absIndexPos);
                    long phyOffsetRead = this.mappedByteBuffer.getLong(absIndexPos + 4);

                    long timeDiff = (long) this.mappedByteBuffer.getInt(absIndexPos + 4 + 8);
                    int prevIndexRead = this.mappedByteBuffer.getInt(absIndexPos + 4 + 8 + 4);
                    // 如果存储的时间差小于0，则直接结束；
                    // 如果hashcode匹配并且消息存储时间介于待查找时间start、end之间则将消息物理偏移量加入到phyOffsets,并验证条目
                    // 前一个Index索引，如果大于等于1并且小于Index条目数，则继续查找，否则结束整个查找。
                    if (timeDiff < 0) {
                        break;
                    }

                    timeDiff *= 1000L;

                    long timeRead = this.indexHeader.getBeginTimestamp() + timeDiff;
                    boolean timeMatched = (timeRead >= begin) && (timeRead <= end);

                    if (keyHash == keyHashRead && timeMatched) {
                        phyOffsets.add(phyOffsetRead);
                    }

                    if (prevIndexRead <= invalidIndex
                        || prevIndexRead > this.indexHeader.getIndexCount()
                        || prevIndexRead == nextIndexToRead || timeRead < begin) {
                        break;
                    }

                    nextIndexToRead = prevIndexRead;
                }
            }
        } catch (Exception e) {
            log.error("selectPhyOffset exception ", e);
        } finally {
            if (fileLock != null) {
                try {
                    fileLock.release();
                } catch (IOException e) {
                    log.error("Failed to release the lock", e);
                }
            }

            this.mappedFile.release();
        }
    }
}