rocketmq源码解析之管理请求查询消费时间②
说在前面java
管理请求 QUERY_CONSUME_TIME_SPAN 查询消费时间apache
源码解析缓存
接上篇微信
按topic、queueId、consumerOffset查询最大时间，往上返回到这个方法org.apache.rocketmq.store.DefaultMessageStore#getMessageStoreTimeStampapp
@Override
public long getMessageStoreTimeStamp(String topic, int queueId, long consumeQueueOffset) {
// 按topic和queueId查询到消费队列=》
ConsumeQueue logicQueue = this.findConsumeQueue(topic, queueId);
if (logicQueue != null) {
// 按消费者的offset查询存储时间所在的buffer=》
SelectMappedBufferResult result = logicQueue.getIndexBuffer(consumeQueueOffset);
// =》
return getStoreTime(result);
}
return -1;
}
按topic和queueId查询到消费队列，进入这个方法org.apache.rocketmq.store.DefaultMessageStore#findConsumeQueueide
public ConsumeQueue findConsumeQueue(String topic, int queueId) {
// 找到topic的全部消息队列
ConcurrentMap<Integer, ConsumeQueue> map = consumeQueueTable.get(topic);
if (null == map) {
ConcurrentMap<Integer, ConsumeQueue> newMap = new ConcurrentHashMap<Integer, ConsumeQueue>(128);
ConcurrentMap<Integer, ConsumeQueue> oldMap = consumeQueueTable.putIfAbsent(topic, newMap);
if (oldMap != null) {
map = oldMap;
} else {
map = newMap;
}
}
// 按queue id查找消费者队列
ConsumeQueue logic = map.get(queueId);
if (null == logic) {
ConsumeQueue newLogic = new ConsumeQueue(
topic,
queueId,
// 消费者队列存储地址 user.home/store/consumequeue
StorePathConfigHelper.getStorePathConsumeQueue(this.messageStoreConfig.getStorePathRootDir()),
// 每一个文件存储默认30W
this.getMessageStoreConfig().getMapedFileSizeConsumeQueue(),
this);
ConsumeQueue oldLogic = map.putIfAbsent(queueId, newLogic);
if (oldLogic != null) {
logic = oldLogic;
} else {
logic = newLogic;
}
}
return logic;
}
按消费者的offset查询存储时间所在的buffer，往上返回到这个方法org.apache.rocketmq.store.ConsumeQueue#getIndexBufferthis
public SelectMappedBufferResult getIndexBuffer(final long startIndex) {
int mappedFileSize = this.mappedFileSize;
// 获取最小的物理offset
long offset = startIndex * CQ_STORE_UNIT_SIZE;
if (offset >= this.getMinLogicOffset()) {
// 根据offset查询映射文件 =》
MappedFile mappedFile = this.mappedFileQueue.findMappedFileByOffset(offset);
if (mappedFile != null) {
SelectMappedBufferResult result = mappedFile.selectMappedBuffer((int) (offset % mappedFileSize));
return result;
}
}
return null;
}
根据offset查询映射文件，进入这个方法org.apache.rocketmq.store.MappedFileQueue#findMappedFileByOffset(long)code
public MappedFile findMappedFileByOffset(final long offset) {
// =》
return findMappedFileByOffset(offset, false);
}
进入这个方法org.apache.rocketmq.store.MappedFileQueue#findMappedFileByOffset(long, boolean)blog
public MappedFile findMappedFileByOffset(final long offset, final boolean returnFirstOnNotFound) {
try {
// 获取队列中第一个映射文件
MappedFile firstMappedFile = this.getFirstMappedFile();
// 获取队列中最后一个映射文件
MappedFile lastMappedFile = this.getLastMappedFile();
if (firstMappedFile != null && lastMappedFile != null) {
// 若是offset不在索引文件的offset范围内
if (offset < firstMappedFile.getFileFromOffset() || offset >= lastMappedFile.getFileFromOffset() + this.mappedFileSize) {
LOG_ERROR.warn("Offset not matched. Request offset: {}, firstOffset: {}, lastOffset: {}, mappedFileSize: {}, mappedFiles count: {}",
offset,
firstMappedFile.getFileFromOffset(),
lastMappedFile.getFileFromOffset() + this.mappedFileSize,
this.mappedFileSize,
this.mappedFiles.size());
} else {
// 找到映射文件在队列中的索引位置
int index = (int) ((offset / this.mappedFileSize) - (firstMappedFile.getFileFromOffset() / this.mappedFileSize));
MappedFile targetFile = null;
try {
// 获取索引文件
targetFile = this.mappedFiles.get(index);
} catch (Exception ignored) {
}
// offset在目标文件的起始offset和结束offset范围内
if (targetFile != null && offset >= targetFile.getFileFromOffset()
&& offset < targetFile.getFileFromOffset() + this.mappedFileSize) {
return targetFile;
}
// 若是按索引在队列中找不到映射文件就遍历队列查找映射文件
for (MappedFile tmpMappedFile : this.mappedFiles) {
if (offset >= tmpMappedFile.getFileFromOffset()
&& offset < tmpMappedFile.getFileFromOffset() + this.mappedFileSize) {
return tmpMappedFile;
}
}
}
// 若是offset=0获取队列中第一个映射文件，我的感受这个逻辑是否放在前面判断更为合理，仍是放在这里另有深意
if (returnFirstOnNotFound) {
return firstMappedFile;
}
}
} catch (Exception e) {
log.error("findMappedFileByOffset Exception", e);
}
return null;
}
往上返回到这个方法org.apache.rocketmq.store.DefaultMessageStore#getStoreTime索引
private long getStoreTime(SelectMappedBufferResult result) {
if (result != null) {
try {
final long phyOffset = result.getByteBuffer().getLong();
final int size = result.getByteBuffer().getInt();
// 根据SelectMappedBufferResult的offset和大小查找存储时间=》
long storeTime = this.getCommitLog().pickupStoreTimestamp(phyOffset, size);
return storeTime;
} catch (Exception e) {
} finally {
result.release();
}
}
return -1;
}
根据SelectMappedBufferResult的offset和大小查找存储时间，进入这个方法org.apache.rocketmq.store.CommitLog#pickupStoreTimestamp
public long pickupStoreTimestamp(final long offset, final int size) {
if (offset >= this.getMinOffset()) {
// =》
SelectMappedBufferResult result = this.getMessage(offset, size);
if (null != result) {
try {
// 获取消息存储时间
return result.getByteBuffer().getLong(MessageDecoder.MESSAGE_STORE_TIMESTAMP_POSTION);
} finally {
result.release();
}
}
}
return -1;
}
进入这个方法org.apache.rocketmq.store.CommitLog#getMessage
public SelectMappedBufferResult getMessage(final long offset, final int size) {
int mappedFileSize = this.defaultMessageStore.getMessageStoreConfig().getMapedFileSizeCommitLog();
// 根据offset找到映射文件 =》
MappedFile mappedFile = this.mappedFileQueue.findMappedFileByOffset(offset, offset == 0);
if (mappedFile != null) {
int pos = (int) (offset % mappedFileSize);
return mappedFile.selectMappedBuffer(pos, size);
}
return null;
}
根据offset找到映射文件，进入这个方法org.apache.rocketmq.store.MappedFileQueue#findMappedFileByOffset(long, boolean)
public MappedFile findMappedFileByOffset(final long offset, final boolean returnFirstOnNotFound) {
try {
// 获取队列中第一个映射文件
MappedFile firstMappedFile = this.getFirstMappedFile();
// 获取队列中最后一个映射文件
MappedFile lastMappedFile = this.getLastMappedFile();
if (firstMappedFile != null && lastMappedFile != null) {
// 若是offset不在索引文件的offset范围内
if (offset < firstMappedFile.getFileFromOffset() || offset >= lastMappedFile.getFileFromOffset() + this.mappedFileSize) {
LOG_ERROR.warn("Offset not matched. Request offset: {}, firstOffset: {}, lastOffset: {}, mappedFileSize: {}, mappedFiles count: {}",
offset,
firstMappedFile.getFileFromOffset(),
lastMappedFile.getFileFromOffset() + this.mappedFileSize,
this.mappedFileSize,
this.mappedFiles.size());
} else {
// 找到映射文件在队列中的索引位置
int index = (int) ((offset / this.mappedFileSize) - (firstMappedFile.getFileFromOffset() / this.mappedFileSize));
MappedFile targetFile = null;
try {
// 获取索引文件
targetFile = this.mappedFiles.get(index);
} catch (Exception ignored) {
}
// offset在目标文件的起始offset和结束offset范围内
if (targetFile != null && offset >= targetFile.getFileFromOffset()
&& offset < targetFile.getFileFromOffset() + this.mappedFileSize) {
return targetFile;
}
// 若是按索引在队列中找不到映射文件就遍历队列查找映射文件
for (MappedFile tmpMappedFile : this.mappedFiles) {
if (offset >= tmpMappedFile.getFileFromOffset()
&& offset < tmpMappedFile.getFileFromOffset() + this.mappedFileSize) {
return tmpMappedFile;
}
}
}
// 若是offset=0获取队列中第一个映射文件，我的感受这个逻辑是否放在前面判断更为合理，仍是放在这里另有深意
if (returnFirstOnNotFound) {
return firstMappedFile;
}
}
} catch (Exception e) {
log.error("findMappedFileByOffset Exception", e);
}
return null;
}
按consumerGroup、topic、queueId查询consumerOffset，往上返回到这个方法org.apache.rocketmq.broker.offset.ConsumerOffsetManager#queryOffset(java.lang.String, java.lang.String, int)
public long queryOffset(final String group, final String topic, final int queueId) {
// topic@group 从本地offset缓存中查询
String key = topic + TOPIC_GROUP_SEPARATOR + group;
ConcurrentMap<Integer, Long> map = this.offsetTable.get(key);
if (null != map) {
Long offset = map.get(queueId);
if (offset != null)
return offset;
}
return -1;
}
按topic、queueId、consumerOffset查询消费时间，往上返回到这个方法org.apache.rocketmq.store.DefaultMessageStore#getMessageStoreTimeStamp
@Override
public long getMessageStoreTimeStamp(String topic, int queueId, long consumeQueueOffset) {
// 按topic和queueId查询到消费队列=》
ConsumeQueue logicQueue = this.findConsumeQueue(topic, queueId);
if (logicQueue != null) {
// 按消费者的offset查询存储时间所在的buffer=》
SelectMappedBufferResult result = logicQueue.getIndexBuffer(consumeQueueOffset);
// =》
return getStoreTime(result);
}
return -1;
}
按topic和queueId查询到消费队列，进入这个方法org.apache.rocketmq.store.DefaultMessageStore#findConsumeQueue
public ConsumeQueue findConsumeQueue(String topic, int queueId) {
// 找到topic的全部消息队列
ConcurrentMap<Integer, ConsumeQueue> map = consumeQueueTable.get(topic);
if (null == map) {
ConcurrentMap<Integer, ConsumeQueue> newMap = new ConcurrentHashMap<Integer, ConsumeQueue>(128);
ConcurrentMap<Integer, ConsumeQueue> oldMap = consumeQueueTable.putIfAbsent(topic, newMap);
if (oldMap != null) {
map = oldMap;
} else {
map = newMap;
}
}
// 按queue id查找消费者队列
ConsumeQueue logic = map.get(queueId);
if (null == logic) {
ConsumeQueue newLogic = new ConsumeQueue(
topic,
queueId,
// 消费者队列存储地址 user.home/store/consumequeue
StorePathConfigHelper.getStorePathConsumeQueue(this.messageStoreConfig.getStorePathRootDir()),
// 每一个文件存储默认30W
this.getMessageStoreConfig().getMapedFileSizeConsumeQueue(),
this);
ConsumeQueue oldLogic = map.putIfAbsent(queueId, newLogic);
if (oldLogic != null) {
logic = oldLogic;
} else {
logic = newLogic;
}
}
return logic;
}
按消费者的offset查询存储时间所在的buffer，往上返回到这个方法org.apache.rocketmq.store.ConsumeQueue#getIndexBuffer
public SelectMappedBufferResult getIndexBuffer(final long startIndex) {
int mappedFileSize = this.mappedFileSize;
// 获取最小的物理offset
long offset = startIndex * CQ_STORE_UNIT_SIZE;
if (offset >= this.getMinLogicOffset()) {
// 根据offset查询映射文件 =》
MappedFile mappedFile = this.mappedFileQueue.findMappedFileByOffset(offset);
if (mappedFile != null) {
SelectMappedBufferResult result = mappedFile.selectMappedBuffer((int) (offset % mappedFileSize));
return result;
}
}
return null;
}
根据offset查询映射文件，进入这个方法org.apache.rocketmq.store.MappedFileQueue#findMappedFileByOffset(long)
public MappedFile findMappedFileByOffset(final long offset) {
// =》
return findMappedFileByOffset(offset, false);
}
进入这个方法org.apache.rocketmq.store.MappedFileQueue#findMappedFileByOffset(long, boolean)
public MappedFile findMappedFileByOffset(final long offset, final boolean returnFirstOnNotFound) {
try {
// 获取队列中第一个映射文件
MappedFile firstMappedFile = this.getFirstMappedFile();
// 获取队列中最后一个映射文件
MappedFile lastMappedFile = this.getLastMappedFile();
if (firstMappedFile != null && lastMappedFile != null) {
// 若是offset不在索引文件的offset范围内
if (offset < firstMappedFile.getFileFromOffset() || offset >= lastMappedFile.getFileFromOffset() + this.mappedFileSize) {
LOG_ERROR.warn("Offset not matched. Request offset: {}, firstOffset: {}, lastOffset: {}, mappedFileSize: {}, mappedFiles count: {}",
offset,
firstMappedFile.getFileFromOffset(),
lastMappedFile.getFileFromOffset() + this.mappedFileSize,
this.mappedFileSize,
this.mappedFiles.size());
} else {
// 找到映射文件在队列中的索引位置
int index = (int) ((offset / this.mappedFileSize) - (firstMappedFile.getFileFromOffset() / this.mappedFileSize));
MappedFile targetFile = null;
try {
// 获取索引文件
targetFile = this.mappedFiles.get(index);
} catch (Exception ignored) {
}
// offset在目标文件的起始offset和结束offset范围内
if (targetFile != null && offset >= targetFile.getFileFromOffset()
&& offset < targetFile.getFileFromOffset() + this.mappedFileSize) {
return targetFile;
}
// 若是按索引在队列中找不到映射文件就遍历队列查找映射文件
for (MappedFile tmpMappedFile : this.mappedFiles) {
if (offset >= tmpMappedFile.getFileFromOffset()
&& offset < tmpMappedFile.getFileFromOffset() + this.mappedFileSize) {
return tmpMappedFile;
}
}
}
// 若是offset=0获取队列中第一个映射文件，我的感受这个逻辑是否放在前面判断更为合理，仍是放在这里另有深意
if (returnFirstOnNotFound) {
return firstMappedFile;
}
}
} catch (Exception e) {
log.error("findMappedFileByOffset Exception", e);
}
return null;
}
往上返回到这个方法org.apache.rocketmq.store.DefaultMessageStore#getStoreTime
private long getStoreTime(SelectMappedBufferResult result) {
if (result != null) {
try {
final long phyOffset = result.getByteBuffer().getLong();
final int size = result.getByteBuffer().getInt();
// 根据SelectMappedBufferResult的offset和大小查找存储时间=》
long storeTime = this.getCommitLog().pickupStoreTimestamp(phyOffset, size);
return storeTime;
} catch (Exception e) {
} finally {
result.release();
}
}
return -1;
}
根据SelectMappedBufferResult的offset和大小查找存储时间，进入这个方法org.apache.rocketmq.store.CommitLog#pickupStoreTimestamp
public long pickupStoreTimestamp(final long offset, final int size) {
if (offset >= this.getMinOffset()) {
// =》
SelectMappedBufferResult result = this.getMessage(offset, size);
if (null != result) {
try {
// 获取消息存储时间
return result.getByteBuffer().getLong(MessageDecoder.MESSAGE_STORE_TIMESTAMP_POSTION);
} finally {
result.release();
}
}
}
return -1;
}
进入这个方法org.apache.rocketmq.store.CommitLog#getMessage
public SelectMappedBufferResult getMessage(final long offset, final int size) {
int mappedFileSize = this.defaultMessageStore.getMessageStoreConfig().getMapedFileSizeCommitLog();
// 根据offset找到映射文件 =》
MappedFile mappedFile = this.mappedFileQueue.findMappedFileByOffset(offset, offset == 0);
if (mappedFile != null) {
int pos = (int) (offset % mappedFileSize);
return mappedFile.selectMappedBuffer(pos, size);
}
return null;
}
根据offset找到映射文件，进入这个方法org.apache.rocketmq.store.MappedFileQueue#findMappedFileByOffset(long, boolean)
public MappedFile findMappedFileByOffset(final long offset, final boolean returnFirstOnNotFound) {
try {
// 获取队列中第一个映射文件
MappedFile firstMappedFile = this.getFirstMappedFile();
// 获取队列中最后一个映射文件
MappedFile lastMappedFile = this.getLastMappedFile();
if (firstMappedFile != null && lastMappedFile != null) {
// 若是offset不在索引文件的offset范围内
if (offset < firstMappedFile.getFileFromOffset() || offset >= lastMappedFile.getFileFromOffset() + this.mappedFileSize) {
LOG_ERROR.warn("Offset not matched. Request offset: {}, firstOffset: {}, lastOffset: {}, mappedFileSize: {}, mappedFiles count: {}",
offset,
firstMappedFile.getFileFromOffset(),
lastMappedFile.getFileFromOffset() + this.mappedFileSize,
this.mappedFileSize,
this.mappedFiles.size());
} else {
// 找到映射文件在队列中的索引位置
int index = (int) ((offset / this.mappedFileSize) - (firstMappedFile.getFileFromOffset() / this.mappedFileSize));
MappedFile targetFile = null;
try {
// 获取索引文件
targetFile = this.mappedFiles.get(index);
} catch (Exception ignored) {
}
// offset在目标文件的起始offset和结束offset范围内
if (targetFile != null && offset >= targetFile.getFileFromOffset()
&& offset < targetFile.getFileFromOffset() + this.mappedFileSize) {
return targetFile;
}
// 若是按索引在队列中找不到映射文件就遍历队列查找映射文件
for (MappedFile tmpMappedFile : this.mappedFiles) {
if (offset >= tmpMappedFile.getFileFromOffset()
&& offset < tmpMappedFile.getFileFromOffset() + this.mappedFileSize) {
return tmpMappedFile;
}
}
}
// 若是offset=0获取队列中第一个映射文件，我的感受这个逻辑是否放在前面判断更为合理，仍是放在这里另有深意
if (returnFirstOnNotFound) {
return firstMappedFile;
}
}
} catch (Exception e) {
log.error("findMappedFileByOffset Exception", e);
}
return null;
}
接下篇
说在最后
本次解析仅表明我的观点，仅供参考。
加入技术微信群
钉钉技术群
相关文章
- 1. rocketmq源码解析之管理请求查询消费时间③
- 2. rocketmq源码解析之管理请求查询消费时间①
- 3. rocketmq源码解析之管理请求查询消费队列
- 4. rocketmq源码解析之管理请求直接消费消息
- 5. rocketmq源码解析之管理请求clone消费组的offset
- 6. rocketmq源码解析之请求处理直接消费消息
- 7. rocketmq源码解析之管理请求查询topic被哪些消费者消费
- 8. rocketmq源码解析之管理请求清除过时的消费队列
- 9. rocketmq源码解析之管理请求获取消费者运行时信息
- 10. rocketmq源码解析之管理请求之查询broker全部消费组状态
- 更多相关文章...
- • 发送ICMP时间戳请求 - TCP/IP教程
- • MySQL子查询详解 - MySQL教程
- • 互联网组织的未来：剖析GitHub员工的任性之源
- • Scala 中文乱码解决
相关标签/搜索
每日一句
-
每一个你不满意的现在，都有一个你没有努力的曾经。
最新文章
- 1. 字节跳动21届秋招运营两轮面试经验分享
- 2. Java 3 年，25K 多吗？
- 3. mysql安装部署
- 4. web前端开发中父链和子链方式实现通信
- 5. 3.1.6 spark体系之分布式计算-scala编程-scala中trait特性
- 6. dataframe2
- 7. ThinkFree在线
- 8. 在线画图
- 9. devtools热部署
- 10. 编译和链接
欢迎关注本站公众号,获取更多信息
相关文章
- 1. rocketmq源码解析之管理请求查询消费时间③
- 2. rocketmq源码解析之管理请求查询消费时间①
- 3. rocketmq源码解析之管理请求查询消费队列
- 4. rocketmq源码解析之管理请求直接消费消息
- 5. rocketmq源码解析之管理请求clone消费组的offset
- 6. rocketmq源码解析之请求处理直接消费消息
- 7. rocketmq源码解析之管理请求查询topic被哪些消费者消费
- 8. rocketmq源码解析之管理请求清除过时的消费队列
- 9. rocketmq源码解析之管理请求获取消费者运行时信息
- 10. rocketmq源码解析之管理请求之查询broker全部消费组状态