RocketMQ的消息查询

按照MessageId查询消息

MsgId 总共 16 字节，包含消息存储主机地址（ip/port），消息 Commit Log offset。从 MsgId 中解析出 Broker 的地址和 Commit Log 的偏移地址，然后按照存储格式所在位置将消息 buffer 解析成一个完整的消息。
在RocketMQ中具体做法是：Client端从MessageId中解析出Broker的地址（IP地址和端口）和Commit Log的偏移地址后封装成一个RPC请求后，通过Remoting通信层发送（业务请求码：
VIEW_MESSAGE_BY_ID）。Broker使用QueryMessageProcessor，使用请求中的 commitLog offset和 size 去 commitLog 中找到真正的记录并解析成一个完整的消息返回。

按照Message Key查询消息
“按照Message Key查询消息”，主要是基于RocketMQ的IndexFile索引文件来实现的。RocketMQ的索引文件逻辑结构，类似JDK中HashMap的实现。索引文件的具体结构如下：

1.根据查询的 key 的 hashcode%slotNum 得到具体的槽的位置（slotNum 是一个索引文件里面包含的最大槽的数目， 例如图中所示 slotNum=5000000）。
2.根据 slotValue（slot 位置对应的值）查找到索引项列表的最后一项（倒序排列，slotValue 总是指向最新的一个索引项）。
3.遍历索引项列表返回查询时间范围内的结果集（默认一次最大返回的 32 条记录）
4.Hash 冲突；
第一种，key 的 hash 值不同但模数相同，此时查询的时候会再比较一次 key 的 hash 值（每个索引项保存了 key 的 hash 值），过滤掉 hash 值不相等的项。
第二种，hash 值相等但 key 不等， 出于性能的考虑冲突的检测放到客户端处理（key 的原始值是存储在消息文件中的，避免对数据文件的解析）， 客户端比较一次消息体的 key 是否相同。
5.存储；为了节省空间索引项中存储的时间是时间差值（存储时间-开始时间，开始时间存储在索引文件头中），整个索引文件是定长的，结构也是固定的。

import org.apache.rocketmq.client.consumer.DefaultMQPullConsumer; 
import org.apache.rocketmq.client.exception.MQBrokerException; 
import org.apache.rocketmq.client.exception.MQClientException; 
import org.apache.rocketmq.common.message.MessageExt; 
import org.apache.rocketmq.remoting.exception.RemotingException; 

public class QueryingMessageDemo { 
    public static void main(String[] args) throws InterruptedException, RemotingException, MQClientException, MQBrokerException { 
        DefaultMQPullConsumer consumer = new DefaultMQPullConsumer("consumer_grp_09_01"); 
        
        consumer.setNamesrvAddr("node1:9876"); 
        consumer.start(); 
        
        MessageExt message = consumer.viewMessage("tp_demo_08", "0A4E00A7178878308DB150A780BB0000"); 
        System.out.println(message); 
        System.out.println(message.getMsgId()); 
        consumer.shutdown(); 
    } 
}