1. Commit Log

2. Index

2.1. IndexHeader

// IndexHeader.java    public static final int INDEX_HEADER_SIZE = 40;    private static int beginTimestampIndex = 0;    private static int endTimestampIndex = 8;    private static int beginPhyoffsetIndex = 16;    private static int endPhyoffsetIndex = 24;    private static int hashSlotcountIndex = 32;    private static int indexCountIndex = 36;    private final ByteBuffer byteBuffer;    private AtomicLong beginTimestamp = new AtomicLong(0);    private AtomicLong endTimestamp = new AtomicLong(0);    private AtomicLong beginPhyOffset = new AtomicLong(0);    private AtomicLong endPhyOffset = new AtomicLong(0);    private AtomicInteger hashSlotCount = new AtomicInteger(0);    private AtomicInteger indexCount = new AtomicInteger(1);    public IndexHeader(final ByteBuffer byteBuffer) {        this.byteBuffer = byteBuffer;    }    public void load() {        this.beginTimestamp.set(byteBuffer.getLong(beginTimestampIndex)); // CCC beginTimestamp  8位long类型，索引文件构建第一个索引的消息落在broker的时间        this.endTimestamp.set(byteBuffer.getLong(endTimestampIndex));     // CCC endTimestamp    8位long类型，索引文件构建最后一个索引消息落broker时间        this.beginPhyOffset.set(byteBuffer.getLong(beginPhyoffsetIndex)); // CCC beginPhyOffset  8位long类型，索引文件构建第一个索引的消息commitLog偏移量        this.endPhyOffset.set(byteBuffer.getLong(endPhyoffsetIndex));     // CCC endPhyOffset    8位long类型，索引文件构建最后一个索引消息commitLog偏移量        this.hashSlotCount.set(byteBuffer.getInt(hashSlotcountIndex));    // CCC hashSlotCount   4位int类型，构建索引占用的槽位数(这个值貌似没有具体作用)        this.indexCount.set(byteBuffer.getInt(indexCountIndex));          // CCC indexCount      4位int类型，索引文件中构建的索引个数        if (this.indexCount.get() <= 0) {            this.indexCount.set(1);        }    }    public void updateByteBuffer() {        this.byteBuffer.putLong(beginTimestampIndex, this.beginTimestamp.get());        this.byteBuffer.putLong(endTimestampIndex, this.endTimestamp.get());        this.byteBuffer.putLong(beginPhyoffsetIndex, this.beginPhyOffset.get());        this.byteBuffer.putLong(endPhyoffsetIndex, this.endPhyOffset.get());        this.byteBuffer.putInt(hashSlotcountIndex, this.hashSlotCount.get());        this.byteBuffer.putInt(indexCountIndex, this.indexCount.get());    }// ---- getter/setter ----// ...

2.2. IndexFile

// IndexFile.java    /* CCC 写index文件 */    public boolean putKey(final String key, final long phyOffset, final long storeTimestamp) {        if (this.indexHeader.getIndexCount() < this.indexNum) {            int keyHash = indexKeyHashMethod(key);            int slotPos = keyHash % this.hashSlotNum;            int absSlotPos = IndexHeader.INDEX_HEADER_SIZE + slotPos * hashSlotSize;            FileLock fileLock = null;            try {                int slotValue = this.mappedByteBuffer.getInt(absSlotPos); // CCC 获取当前slot段中的值                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 = // CCC 索引段位置                    IndexHeader.INDEX_HEADER_SIZE + this.hashSlotNum * hashSlotSize                        + this.indexHeader.getIndexCount() * indexSize;                this.mappedByteBuffer.putInt(absIndexPos, keyHash);                       // CCC  keyHash           4位int值，存储的是key的hash值                this.mappedByteBuffer.putLong(absIndexPos + 4, phyOffset);          // CCC  commitlog offset  8位long值，存储的是消息在commitlog的物理偏移量phyOffset                this.mappedByteBuffer.putInt(absIndexPos + 4 + 8, (int) timeDiff);  // CCC  timediff          4位int值，存储了当前消息跟索引文件中第一个消息在broker落地的时间差                this.mappedByteBuffer.putInt(absIndexPos + 4 + 8 + 4, slotValue);   // CCC  slotValue         4位int值，无hash冲突，slotValue=0；若hash冲突存储上一个消息的slotValue槽位，计算索引地址                this.mappedByteBuffer.putInt(absSlotPos, this.indexHeader.getIndexCount());// CCC absSlotPos        4位int值，slot槽位存储消息个数索引                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);            } 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;    }    public int indexKeyHashMethod(final String key) {        int keyHash = key.hashCode();        int keyHashPositive = Math.abs(keyHash);        if (keyHashPositive < 0)            keyHashPositive = 0;        return keyHashPositive;    }    /* CCC 读Index文件 */    public void selectPhyOffset(final List
    
      phyOffsets, final String key, final int maxNum,        final long begin, final long end, boolean lock) {        if (this.mappedFile.hold()) {            int keyHash = indexKeyHashMethod(key);            int slotPos = keyHash % this.hashSlotNum;            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;                // }                if (slotValue <= invalidIndex || slotValue > this.indexHeader.getIndexCount()                    || this.indexHeader.getIndexCount() <= 1) {                } else {                    for (int nextIndexToRead = slotValue; ; ) {                        if (phyOffsets.size() >= maxNum) {                            break;                        }                        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);                        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();            }        }    }
    

3. 参考资料

略。