为何RocketMQ的ScheduleMessageService用tagsCode计算投递时间?
这是RocketMQ为优化延迟消息存储与处理性能采用的字段复用设计,核心逻辑如下:
- 延迟消息首次写入CommitLog时,不会额外开辟字段存储目标投递时间,而是把计算好的投递时间戳直接存入原本用于存储消息标签哈希值的
tagsCode字段。此时tagsCode的语义已从标签哈希变为投递时间的载体。 - 当
cqUnit.isTagsCodeValid()返回true时,说明当前tagsCode已被替换为合法的投递时间戳,直接使用即可,无需再去CommitLog中读取消息存储时间重新计算,避免了额外IO操作,提升处理效率。 - 只有当
tagsCode无效(比如非延迟消息、处理异常导致字段未被正确替换等场景),才会通过pickupStoreTimestamp从CommitLog中读取消息的存储时间msgStoreTime,再结合延迟等级delayLevel调用computeDeliverTimestamp计算投递时间。
相关核心代码:
CqUnit cqUnit = bufferCQ.next(); long offsetPy = cqUnit.getPos(); int sizePy = cqUnit.getSize(); long tagsCode = cqUnit.getTagsCode(); if (!cqUnit.isTagsCodeValid()) { long msgStoreTime = ScheduleMessageService.this.brokerController.getMessageStore().getCommitLog().pickupStoreTimestamp(offsetPy, sizePy); tagsCode = computeDeliverTimestamp(delayLevel, msgStoreTime); } long now = System.currentTimeMillis(); long deliverTimestamp = this.correctDeliverTimestamp(now, tagsCode);
内容的提问来源于stack exchange,提问作者zhangzhitong
相关产品推荐
相关产品推荐

