标签：ash map bsp turn key ble roc read nal
如果一个broker正常退出，是会删除abort文件的。那么启动broker的时候发现abort文件还存在，那么说明上次是异常终止，会进入到commit-log的recoverAbnormally逻辑里面，因为所有其他的信息都是从commit-log获取到的，所以追根溯源只能从commit-log开始着手。
public void recoverAbnormally(long maxPhyOffsetOfConsumeQueue) {
// recover by the minimum time stamp
boolean checkCRCOnRecover = this.defaultMessageStore.getMessageStoreConfig().isCheckCRCOnRecover();
final List<MappedFile> mappedFiles = this.mappedFileQueue.getMappedFiles();
if (!mappedFiles.isEmpty()) {
// Looking beginning to recover from which file
int index = mappedFiles.size() - 1;
MappedFile mappedFile = null;
for (; index >= 0; index--) {
mappedFile = mappedFiles.get(index);
if (this.isMappedFileMatchedRecover(mappedFile)) {
log.info("recover from this mapped file " + mappedFile.getFileName());
break;
}
}
从最新的mapped文件开始找，直到发现一个有效的mappedfile，对这个mappedfile文件执行操作：
ByteBuffer byteBuffer = mappedFile.sliceByteBuffer();
while (true) {
DispatchRequest dispatchRequest = this.checkMessageAndReturnSize(byteBuffer, checkCRCOnRecover);
int size = dispatchRequest.getMsgSize();
if (dispatchRequest.isSuccess()) {
// Normal data
if (size > 0) {
mappedFileOffset += size;
if (this.defaultMessageStore.getMessageStoreConfig().isDuplicationEnable()) {
if (dispatchRequest.getCommitLogOffset() < this.defaultMessageStore.getConfirmOffset()) {
this.defaultMessageStore.doDispatch(dispatchRequest);
}
} else {
this.defaultMessageStore.doDispatch(dispatchRequest);
}
}
checkMessageAndReturnSize的逻辑：
public DispatchRequest checkMessageAndReturnSize(java.nio.ByteBuffer byteBuffer, final boolean checkCRC,
final boolean readBody) {
try {
// 1 TOTAL SIZE
int totalSize = byteBuffer.getInt();
通过不变遍历这个最新的mappedfile的bytebuffer，把他的每一条消息取出来，并且消息一个字段就是total-size，累计求和就可以拿到这个mappedfile上次挂掉的时候到底写到哪了。对于其他的mappedfile默认认为都是写满的。
processOffset += mappedFileOffset;
this.mappedFileQueue.setFlushedWhere(processOffset);
this.mappedFileQueue.setCommittedWhere(processOffset);
this.mappedFileQueue.truncateDirtyFiles(processOffset);
这里已经把commit-log完全恢复了。接下来就是恢复consumequeue、indexfile了。前一节提到生成消息的时候，其实也会同步更新consumeQueue，并且刷盘。
也就是不管是故障恢复还是正常生成消息，都会涉及commit-log、consumeQueue、indexfile的更新、落盘。只不过后面两个是异步的。
他们都是通过this.defaultMessageStore.doDispatch(dispatchRequest)更新的
另外topicQueueTable完全可以通过consumeQueue得到，因为前者是后者子集：
private void recoverTopicQueueTable() {
HashMap<String/* topic-queueid */, Long/* offset */> table = new HashMap<String, Long>(1024);
long minPhyOffset = this.commitLog.getMinOffset();
for (ConcurrentMap<Integer, ConsumeQueue> maps : this.consumeQueueTable.values()) {
for (ConsumeQueue logic : maps.values()) {
String key = logic.getTopic() + "-" + logic.getQueueId();
table.put(key, logic.getMaxOffsetInQueue());
logic.correctMinOffset(minPhyOffset);
}
}
this.commitLog.setTopicQueueTable(table);
}
rocketmq的broker恢复commit-log的时候如何恢复consumeQueue、indexfile
标签：ash map bsp turn key ble roc read nal
原文地址：https://www.cnblogs.com/notlate/p/12008345.html
