RocketMQ 整合 DLedger 设计技巧与实现原理

if (messageStoreConfig.isEnableDLegerCommitLog()) {
    this.commitLog = new DLedgerCommitLog(this);
} else {
    this.commitLog = new CommitLog(this);
}

if (messageStoreConfig.isEnableDLegerCommitLog()) {
    DLedgerRoleChangeHandler roleChangeHandler = new DLedgerRoleChangeHandler(this, (DefaultMessageStore) messageStore);
    ((DLedgerCommitLog) ((DefaultMessageStore) messageStore).getCommitLog()).getdLedgerServer().getdLedgerLeaderElector(
    ).addRoleChangeHandler(roleChangeHandler);
}

result = result && this.commitLog.load();
result = result && this.loadConsumeQueue();

if (result) {
    this.storeCheckpoint = new StoreCheckpoint(StorePathConfigHelper.getStoreCheckpoint(this.messageStoreConfig.getStorePathRootDir()));
    this.indexService.load(lastExitOK);
    this.recover(lastExitOK);
    log.info("load over, and the max phyoffset = {}", this.getMaxPhyOffset());
}

super(defaultMessageStore);

dLedgerConfig = new DLedgerConfig();
dLedgerConfig.setEnableDiskForceClean(defaultMessageStore.getMessageStoreConfig().isCleanFileForciblyEnable());
dLedgerConfig.setStoreType(DLedgerConfig.FILE);
dLedgerConfig.setSelfId(defaultMessageStore.getMessageStoreConfig().getdLegerSelfId());
dLedgerConfig.setGroup(defaultMessag eStore.getMessageStoreConfig().getdLegerGroup());
dLedgerConfig.setPeers(defaultMessageStore.getMessageStoreConfig().getdLegerPeers());
dLedgerConfig.setStoreBaseDir(defaultMessageStore.getMessageStoreConfig().getStorePathRootDir());
dLedgerConfig.setMappedFileSizeForEntryData(defaultMessageStore.getMessageStoreConfig().getMappedFileSizeCommitLog());
dLedgerConfig.setDeleteWhen(defaultMessageStore.getMessageStoreConfig().getDeleteWhen());
dLedgerConfig.setFileReservedHours(defaultMessageStore.getMessageStoreConfig().getFileReservedTime() + 1);
id = Integer.valueOf(dLedgerConfig.getSelfId().substring(1)) + 1;

dLedgerServer = new DLedgerServer(dLedgerConfig);
dLedgerFileStore = (DLedgerMmapFileStore) dLedgerServer.getdLedgerStore();

DLedgerMmapFileStore.AppendHook appendHook = (entry, buffer, bodyOffset) -> {
    assert bodyOffset == DLedgerEntry.BODY_OFFSET;
    buffer.position(buffer.position() + bodyOffset + MessageDecoder.PHY_POS_POSITION);
    buffer.putLong(entry.getPos() + bodyOffset);
};
dLedgerFileStore.addAppendHook(appendHook);

public boolean load() {
    return super.load();
}

dLedgerFileStore.load();

if (dLedgerFileList.getMappedFiles().size() > 0) {
    dLedgerFileStore.recover();
    dividedCommitlogOffset = dLedgerFileList.getFirstMappedFile().getFileFromOffset();
    MappedFile mappedFile = this.mappedFileQueue.getLastMappedFile();
    if (mappedFile != null) {
        disableDeleteDledger();
    }
    long maxPhyOffset = dLedgerFileList.getMaxWrotePosition();
    if (maxPhyOffsetOfConsumeQueue >= maxPhyOffset) {
        log.warn("[TruncateCQ]maxPhyOffsetOfConsum eQueue({}) >= processOffset({}), truncate dirty logic files", maxPhyOffsetOfConsumeQueue, maxPhyOffset);
        this.defaultMessageStore.truncateDirtyLogicFiles(maxPhyOffset);
    }
    return;
}

isInrecoveringOldCommitlog = true;
super.recoverNormally(maxPhyOffsetOfConsumeQueue);

MappedFile mappedFile = this.mappedFileQueue.getLastMappedFile();
if (mappedFile == null) {
    return;
}

ByteBuffer byteBuffer = mappedFile.sliceByteBuffer();
byteBuffer.position(mappedFile.getWrotePosition());
boolean needWriteMagicCode = true;
byteBuffer.getInt(); //size
int magicCode = byteBuffer.getInt();
if (magicCode == CommitLog.BLANK_MAGIC_CODE) {
    needWriteMagicCode = false;
} else {
    log.info("Recover old commitlog found a illegal magic code = {}", magicCode);
}
dLedgerConfig.setEnableDiskForceClean(false);
dividedCommitlogOffset = mappedFile.getFileFromOffset() + mappedFile.getFileSize();
if (needWriteMagicCode) {
    byteBuffer.position(mappedFile.getWrotePosition());
    byteBuffer.putInt(mappedFile.getFileSize() - mappedFile.getWrotePosition());
    byteBuffer.putInt(BLANK_MAGIC_CODE);
    mappedFile.flush(0);
}
mappedFile.setWrotePosition(mappedFile.getFileSize());
mappedFile.setCommittedPosition(mappedFile.getFileSize());
mappedFile.setFlushedPosition(mappedFile.getFileSize());
dLedgerFileList.getLastMappedFile(dividedCommitlogOffset);

AppendEntryRequest request = new AppendEntryRequest();
request.setGroup(dLedgerConfig.getGroup());
request.setRemoteId(dLedgerServer.getMemberState().getSelfId());
request.setBody(encodeResult.data);
dledgerFuture = (AppendFuture<AppendEntryResponse>) dLedgerServer.handleAppend(request);
if (dledgerFuture.getPos() == -1) {
    return new PutMessageResult(PutMessageStatus.OS_PAGECACHE_BUSY, new AppendMessageResult(AppendMessageStatus.UNKNOWN_ERROR);
}

long wroteOffset = dledgerFuture.getPos() + DLedgerEntry.BODY_OFFSET;
appendResult = new AppendMessageResult(AppendMessageStatus.PUT_OK, wroteOffset, encodeResult.data.length, msgId, System.currentTimeMillis(), queueOffset, elapsedTimeInLock);

public SelectMappedBufferResult getMessage ( final long offset, final int size){
    if (offset < dividedCommitlogOffset) {
        return super.getMessage(offset, size);
    }
    int mappedFileSize = this.dLedgerServer.getdLedgerConfig().getMappedFileSizeForEntryData();
    MmapFile mappedFile = this.dLedgerFileList.findMappedFileByOffset(offset, offset == 0);
    if (mappedFile != null) {
        int pos = (int) (offset % mappedFileSize);
        return convertSbr(mappedFile.selectMappedBuffer(pos, size));
    }
    return null;
}

private void startProcessorByHa(BrokerRole role) {
    if (BrokerRole.SLAVE != role) {
        if (this.transactionalMessageCheckService != null) {
            this.transactionalMessageCheckService.start();
        }
    }
}

private void shutdownProcessorByHa() {
    if (this.transactionalMessageCheckService != null){
        this.transactionalMessageCheckService.shutdown(true);
    }
}

private void handleSlaveSynchronize(BrokerRole role) {
    if (role == BrokerRole.SLAVE) {
        if (null != slaveSyncFuture) {
            slaveSyncFuture.cancel(false);
        }
        this.slaveSynchronize.setMasterAddr(null);
        slaveSyncFuture = this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
                                  @Override
                                  public void run() {
                                      try {
                                          BrokerController.this.slaveSynchronize.syncAll();
                                      } catch (Throwable e) {
                                      }
                                  }
                              }, 1000 * 3, 1000 * 10, TimeUnit.MILLISECONDS
        );
    } else {
        if (null != slaveSyncFuture) {
            slaveSyncFuture.cancel(false);
        }
        this.slaveSynchronize.setMasterAddr(null);
    }
}

public void changeToSlave(int brokerId) {
    log.info("Begin to change to slave brokerName={} brokerId={}", brokerConfig.getBrokerName(), brokerId);
    brokerConfig.setBrokerId(brokerId == 0 ? 1 : brokerId); //TO DO check
    messageStoreConfig.setBrokerRole(BrokerRole.SLAVE);
    try {
        this.messageStore.handleScheduleMessageService(BrokerRole.SLAVE);
    } catch (Throwable t) {
        log.error("[MONITOR] handleScheduleMessageService failed when changing to slave", t);
    }
    try {
        this.shutdownProcessorByHa();
    } catch (Throwable t) {
        log.error(" [MONITOR]shutdownProcessorByHa failed when changing to slave", t);
    }
    handleSlaveSynchronize(BrokerRole.SLAVE);
    try {
        this.registerBrokerAll(true, true, brokerConfig.isForceRegister());
    } catch (Throwable ignored) {
    }
    log.info("Finish to change to slave brokerName = {}brokerId = {}", brokerConfig.getBrokerName(), brokerId);
}

public void changeToMaster(BrokerRole role) {
    if (role == BrokerRole.SLAVE) {
        return;
    }
    handleSlaveSynchronize(role);
    try {
        this.messageStore.handleScheduleMessageService(role);
    } catch (Throwable t) {
        log.error("[MONITOR] handleSchedu leMessageService failed when changing to master", t);
    }
    try {
        this.startProcessorByHa(BrokerRole.SYNC_MASTER);
    } catch (Throwable t) {
        log.error("[MONITOR] startProcessorByHa failed when changing to master", t);
    }
    brokerConfig.setBrokerId(0); //TO DO
    check messageStoreConfig.setBrokerRole(role);
    try {
        this.registerBrokerAll(true, true, brokerConfig.isForceRegister());
    } catch (Throwable ignored) {
    }
}

if (messageStoreConfig.isEnableDLegerCommitLog()) {
    DLedgerRoleChangeHandler roleChangeHandler = new DLedgerRoleChangeHandler(this, (DefaultMessageStore) messageStore);
    ((DLedgerCommitLog) ((DefaultMessageStore) messageStore).getCommitLog()).getdLedgerServer().getdLedgerLeaderElector().addRoleChangeHandler(roleChangeHandler);
}

case CANDIDATE:
    if (messageStore.getMessageStoreConfig().getBrokerRole() != BrokerRole.SLAVE) {
        brokerController.changeToSlave(dLedgerCommitLog.getId());
    }
    break;

case FOLLOWER:
    brokerController.changeToSlave(dLedgerCommitLog.getId());
    break;

case LEADER:
    while (true) {
        if (!dLegerServer.getMemberState().isLeader()) {
            succ = false;
            break;
        }
        if (dLegerServer.getdLedgerStore().getLedgerEndIndex() == -1) {
            break;
        }
        if (dLegerServer.getdLedgerStore().getLedgerEndIndex() == dLegerServer.getdLedgerStore().getCommittedIndex()
                && messageStore.dispatchBehindBytes() == 0) {
            break;
        }
        Thread.sleep(100);
    }
    if (succ) {
        messageStore.recoverTopicQueueTable();
        brokerController.changeToMaster(BrokerRole.SYNC_MASTER);
    }
    break;