dledger 文件存储

文件分为 data 和 index

public DLedgerMmapFileStore(DLedgerConfig dLedgerConfig, MemberState memberState) {
    this.dLedgerConfig = dLedgerConfig;
    this.memberState = memberState;
    // 对应 data
    this.dataFileList = new MmapFileList(dLedgerConfig.getDataStorePath(), dLedgerConfig.getMappedFileSizeForEntryData());
    // 对应 file
    this.indexFileList = new MmapFileList(dLedgerConfig.getIndexStorePath(), dLedgerConfig.getMappedFileSizeForEntryIndex());
    localEntryBuffer = ThreadLocal.withInitial(() -> ByteBuffer.allocate(4 * 1024 * 1024));
    localIndexBuffer = ThreadLocal.withInitial(() -> ByteBuffer.allocate(INDEX_UNIT_SIZE * 2));
    flushDataService = new FlushDataService("DLedgerFlushDataService", logger);
    cleanSpaceService = new CleanSpaceService("DLedgerCleanSpaceService", logger);
}

data 存储单纯的数据

index 每一个 entry 有固定的格式和大小，存储 data 的位置，任期和 index

// io.openmessaging.storage.dledger.store.file.DLedgerMmapFileStore#INDEX_UNIT_SIZE
public static final int INDEX_UNIT_SIZE = 32;

/**
 * dataPos 数据的起始位置
 * entrySize 数据大小
 * magic 魔数
 * nextIndex 当前下标
 * currTerm 当前任期
 * indexBuffer 装 index 的 byte 容器
  */
DLedgerEntryCoder.encodeIndex(dataPos, entrySize, CURRENT_MAGIC, nextIndex, memberState.currTerm(), indexBuffer);

只有 leader 接收写数据 (append 请求)，leader 把数据和 index 写入本地后，push 数据给 follower

关于数据的 push 和接收在 io.openmessaging.storage.dledger.DLedgerEntryPusher 类中

// io.openmessaging.storage.dledger.DLedgerEntryPusher#startup
public void startup() {
    // 如果当前节点为 follower 角色，处理 leader 推送过来的消息
    entryHandler.start();
    // 如果当前节点为 leader 角色，根据 peerWaterMark 完成 Future，触发发送响应给客户端
    quorumAckChecker.start();
    // 如果当前节点为 leader 角色，在 3 节点集群中，则创建 2 个 EntryDispatcher，对应 2 个 follower，push 数据给 follower
    for (EntryDispatcher dispatcher : dispatcherMap.values()) {
        dispatcher.start();
    }
}

dledger 接收请求并处理的数据流

io.openmessaging.storage.dledger.DLedgerRpcNettyService#processRequest
io.openmessaging.storage.dledger.DLedgerRpcNettyService#handleAppend
io.openmessaging.storage.dledger.DLedgerServer#handleAppend
  // 写入本地存储
  io.openmessaging.storage.dledger.store.DLedgerStore#appendAsLeader
  // leader 推送给 followers，异步处理 follower 的确认，更新 peerWaterMark
  io.openmessaging.storage.dledger.DLedgerEntryPusher.EntryDispatcher#doAppendInner
  // 返回异步的 Futrue
  io.openmessaging.storage.dledger.DLedgerEntryPusher#waitAck
  // QuorumAckChecker 根据 peerWaterMark 完成 Future
  io.openmessaging.storage.dledger.DLedgerEntryPusher.QuorumAckChecker#doWork