RocketMQ的消息查询（查询消息）

Posted 2021-12-31 刘Java

RocketMQ支持在客户端按照“Message Id”和“Message Key”这两种维度进行消息查询。

文章目录

• 1 按照MessageId查询消息
• 2 按照Message Key查询消息
• 3 相关查询API方法
• 4 UNIQ_KEY

1 按照MessageId查询消息

RocketMQ中的MessageId的长度总共有16字节，其中包含了消息存储主机地址（IP地址和端口），消息Commit Log offset。“按照MessageId查询消息”在RocketMQ中具体做法是：Client端从MessageId中解析出Broker的地址（IP地址和端口）和Commit Log的偏移地址后封装成一个RPC请求后通过Remoting通信层发送（业务请求码：VIEW_MESSAGE_BY_ID）。Broker端走的是QueryMessageProcessor，读取消息的过程用其中的 commitLog offset 和 size 去 commitLog 中找到真正的记录并解析成一个完整的消息返回。

2 按照Message Key查询消息

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

IndexFile索引文件主要用于为用户提供通过“按照Message Key查询消息”的消息索引查询服务，IndexFile文件的存储位置是：KaTeX parse error: Undefined control sequence: \\store at position 5: HOME\\̲s̲t̲o̲r̲e̲\\indexfileName，文件名fileName是以创建时的时间戳命名的，文件大小是固定的，等于40+500W4+2000W20= 420000040个字节大小。如果消息的properties中设置了UNIQ_KEY这个属性，就用 topic + “#” + UNIQ_KEY的value作为 key 来做写入操作。如果消息设置了KEYS属性（多个KEY以空格分隔），也会用 topic + “#” + KEY 来做索引。也就是说，IndexFile支持通过Topic以及UNIQ_KEY或者KEYS来查询。

其中的索引数据包含了Key Hash/CommitLog Offset/Timestamp/NextIndex offset 这四个字段，一共20 Byte。NextIndex offset 即前面读出来的 slotValue，如果有 hash冲突，就可以用这个字段将所有冲突的索引用链表的方式串起来了。Timestamp记录的是消息storeTimestamp之间的差，并不是一个绝对的时间。整个Index File的结构如图，40 Byte 的Header用于保存一些总的统计信息，4*500W的 Slot Table并不保存真正的索引数据，而是保存每个槽位对应的单向链表的头。20*2000W 是真正的索引数据，即一个 Index File 可以保存 2000W个索引。

按照Message Key查询消息的方式，RocketMQ的具体做法是，主要通过Broker端的QueryMessageProcessor业务处理器来查询，读取消息的过程就是用topic和key找到IndexFile索引文件中的一条记录，根据其中的commitLog offset从CommitLog文件中读取消息的实体内容。

3 相关查询API方法

经常使用的DefaultMQProducer、DefaultMQPushConsumer等，都实现了org.apache.rocketmq.client.MQAdmin接口，这个接口中定义了一个用于查询消息的API方法：

/**
 * 查询消息
 *
 * @param topic  消息主题
 * @param key    消息 key
 * @param maxNum 查询返回的最大消息数
 * @param begin  起始时间
 * @param end    结束时间
 * @return QueryResult，内部可能包含多条消息，默认最多返回64条消息
 */
QueryResult queryMessage(final String topic, final String key, final int maxNum, final long begin,
                         final long end) throws MQClientException, InterruptedException;

/**
 * 根据消息id查询消息
 *
 * @param offsetMsgId SendResult中的offsetMsgId
 * @return 返回单条消息
 */
MessageExt viewMessage(final String offsetMsgId) throws RemotingException, MQBrokerException,
        InterruptedException, MQClientException;

/**
 * 根据消息id查询消息
 *
 * @param topic 消息主题
 * @param msgId SendResult中的offsetMsgId、msgId（Unique Key）都可以
 * @return 返回单条消息
 */
MessageExt viewMessage(String topic,
                       String msgId) throws RemotingException, MQBrokerException, InterruptedException, MQClientException;

4 UNIQ_KEY

除了明确的指定消息中的key，RocketMQ生产者客户端在发送发送消息之前，会自动生成一个UNIQ_KEY，设置到消息的属性中，从逻辑上唯一代表一条消息。用户主动设置的Key以及客户端自动生成的UNIQ_KEY，最终都会被设置到Message对象的properties属性中。UNIQ_KEY的值与发送消息之后返回的SendResult结果中的msgId字段的值是一样的，而SendResult结果中的OffsetMsgId才死真正的Message Id。

由于UNIQ_KEY也是使用IndexFile索引文件来存储的，实际上，viewMessage方法传递UNIQ_KEY参数的时候，内部还是调用的queryMessage方法，它和Message key的查询逻辑是一样的，它们都是使用了RocketMQ的哈希索引机制来完成消息查询，因此可能会有哈希冲突，即可能找到多条消息，但是viewMessage方法传递UNIQ_KEY参数的时候还是只返回单条消息，这仅仅是因为对返回的list只取了第一条消息而已。

UNIQ_KEY从逻辑上唯一代表一条消息，这里的“逻辑”是什么意思呢？实际上可能出现多条消息的UNIQ_KEY一致，而且出现这种情况的原因之一就是生产者发送了重复的消息：消息本来已经被成功发送到服务端并完成持久化，由于网络超时导致服务端对客户端应答失败，此时生产者将再次尝试发送相同的消息。

因为在生成UNIQ_KEY的时候会判断UNIQ_KEY属性是否不为null，如果不为null则不会再次生成，那么如果一个消息重新发送多次的话，虽然在Broker中被存储为多条消息，并且它们的Message Id不同，但是他们的UNIQ_KEY都是一致的，即这些消息在逻辑上都是唯一一条消息。

这里也能看出UNIQ_KEY的作用，即在消费时可以通过判断是否出现了相同的UNIQ_KEY来判断这些消息是否是因为生产者重复发送而导致的重复消息。注意UNIQ_KEY并不能彻底解决重复消费的问题，因为还有其他出现重复消费的可能：比如说消费者消费成功了，还没有返回SUCCESS时服务器挂了，再重启的时候将会再次消费者条消息。又比如说消费者Rebalance的时候，可能出现拉取了消息还没有被成功消费并提交的时候，该队列被分配给了其他消费者，导致多个消费者消费同样的消息。

相关文章：

RocketMQ

如有需要交流，或者文章有误，请直接留言。另外希望点赞、收藏、关注，我将不间断更新各种Java学习博客！