消息查询介绍
3种消息查询⽅式：
- Message Key 查询：消息的key是业务开发在发送消息之前⾃⾏指定的，通常会把具有业务含义，区分度⾼的字段作为消息的key，如⽤户id，订单id等。
- Unique Key查询：除了业务开发明确的指定消息中的key，RocketMQ⽣产者客户端在发送发送消息之前，会⾃动⽣成⼀个UNIQ_KEY，设置到消息的属性中，从逻辑上唯⼀代表⼀条消息。
- Message Id 查询：Message Id 是消息发送后，在Broker端⽣成的，其包含了Broker的地址，和在CommitLog中的偏移信息，并会将Message Id作为发送结果的⼀部分进⾏返回。Message Id中属于精确匹配，可以唯⼀定位⼀条消息，不需要使⽤哈希索引机制，查询效率更⾼。
RocketMQ有意弱化Unique Key与Message Id的区别，对外都称之为Message Id。在通过RocketMQ的命令⾏⼯具或管理平台进⾏查询时，⼆者可以通⽤。在根据Unique Key进⾏查询时，本身是有可能查询到多条消息的，但是查询⼯具会进⾏过滤，只会返回⼀条消息。
业务开发同学在使⽤RocketMQ时，应该养成良好的习惯，在发送/消费消息时，将这些信息记录下来，通常是记录到⽇志⽂件中，以便在出现问题时进⾏排查。
//1 构建消息对象Message
Message msg = new Message();
msg.setTopic("TopicA");
msg.setKeys("Key1");
msg.setBody("message body".getBytes());
try{
//2 发送消息
SendResult result = producer.send(msg);
//3 打印发送结果
System.out.println(result);
}catch (Exception e){
e.printStackTrace();
}
事实上，⽤户主动设置的Key以及客户端⾃动⽣成的Unique Key，最终都会设置到Message对象的
properties属性中，如下图所示
其中：
KEYS：表示⽤户通过setKeys⽅法设置的消息key，
UNIQ_KEY：表示客户端⾃动⽣成的Unique Key。
结果中包含Unique Key和Message Id，如下所示：
SendResult [sendStatus=SEND_OK, msgId=0A1427544F4818B4AAC27DD168880000,
offsetMsgId=0A14275400002A9F00000000001F268E, messageQueue=MessageQueue
[topic=TopicTest, brokerName=broker-a, queueId=2], queueOffset=1173]
其中：
- sendStatus：表示消息发送结果的状态
- msgId：注意这⾥的命名虽然是msgId，但实际上其是Unique Key
- offsetMsgId：Broker返回的Message ID 。在后⽂中，未进⾏特殊说明的情况下，Message ID总
是表示offsetMsgId。 - messageQueue：消息发送到了哪个的队列。
- queueOffset：消息在队列中的偏移量，每次发送到⼀个队列时，offset+1
消息查询⼯具
命令⾏⼯具
管理平台
客户端API
命令⾏⼯具
$ sh bin/mqadmin
The most commonly used mqadmin commands are:
...
queryMsgById 按照Message Id查询消息
queryMsgByKey 按照Key查询消息
queryMsgByUniqueKey 按照UNIQ_KEY查询消息
...
例如，要查询在TopicA中，key为Key-0的消息
这⾥，我们看到输出结果中包含了2条记录。其中：
- Message ID列：这⾥这⼀列的名字显示有问题，实际上其代表的是Unique Key
- QID列：表示队列的ID，注意在RocketMQ中唯⼀地位⼀个队列需要topic+brokerName+queueId。这⾥只显示了queueId，其实并不能知道在哪个Broker上。
- Offset：消息在在队列中的偏移量
在查询到Unique Key之后，我们就可以使⽤另外⼀个命令：queryMsgByUniqueKey，来查询消息的具体内容。
对于消息体的内容，会存储到Message Body Path字段指定到的路径中。可通过cat命令查看(仅适⽤于消息体是字符串)：
指定消费者重新消费：
queryMsgByUniqueKey⼦命令还接收另外两个参数：-g参数⽤于指定消费者组名称，-d参数指定消费者client id。指定了这两个参数之后，消息将由消费者直接消费，⽽不是打印在控制台上。
⾸先，通过consumerStatus命令，查询出消费者组下的client id信息，如：
这⾥显示了消费者组please_rename_unique_group_name下⾯只有⼀个消费者，client id为
10、 20.39.84@20820；
接着我们可以在queryMsgByUniqueKey⼦命令中，添加-g和-d参数，如下所示：
可以看到，这⾥并没有打印出消息内容，取⽽代之的是消息消费的结果。
在内部，主要是分为3个步骤来完成让指定消费者来消费这条消息，如下图所示：
第1步：
命令⾏⼯具给所有Broker发起QUERY_MESSAGE请求查询消息，因为并不知道UNIQ_KEY这条消息在哪个Broker上，且最多只会返回⼀条消息，如果超过1条其他会过滤掉；如果查询不到就直接报错。
第2步：
根据消息中包含了Store Host信息，也就是消息存储在哪个Broker上，接来下命令⾏⼯具会直接给这个Broker发起CONSUME_MESSAGE_DIRECTLY请求，这个请求会携带msgId，group和client id的信息
第3步：
Broker接收到这个请求，查询出消息内容后，主动给消费者发送CONSUME_MESSAGE_DIRECTLY通知请求，注意虽然与第2步使⽤了同⼀个请求码，但不同的是这个请求中包含了消息体的内容，消费者可直接处理。注意：这⾥并不是将消息重新发送到Topic中，否则订阅这个Topic的所有消费者组，都会重新消费这条消息。
管理平台
根据Topic时间范围查询：
按Topic 查询属于范围查询，不推荐使⽤，因为时间范围内消息很多，不具备区分度。查询时，尽可能设置最为精确的时间区间，以便缩⼩查询范围，提⾼速度。最多返回2000条数据。
根据Message Key查询：
按Message Key 查询属于模糊查询，仅适⽤于没有记录 Message ID 但是设置了具有区分度的Message Key的情况。 ⽬前，根据Message Key查询，有⼀个很⼤局限性：不能指定时间范围，且最多返回64条数据。如果⽤户指定的key重复率⽐较⾼的话，就有可能搜不到。
根据Message Id查询：
按Message ID 查询属于精确查询，速度快，精确匹配，只会返回⼀条结果，推荐使⽤。在这⾥，传⼊Unique Key，offsetMsgId都可以。
客户端API
除了通过命令⾏⼯具和管理平台，还可以通过客户端API的⽅式来进⾏查询，这其实是最本质的⽅式，命令⾏⼯具和管理平台的查询功能都是基于此实现。
在org.apache.rocketmq.client.MQAdmin接⼝中，定义了以下⼏个⽅法⽤于消息查询：
常⽤的DefaultMQProducer、DefaultMQPushConsumer等，都实现了此接⼝，因此都具备消息查询的能⼒.
public interface MQAdmin {
...
//msgId参数：仅接收SendResult中的offsetMsgId，返回单条消息
MessageExt viewMessage(final String msgId)
//msgId参数：传⼊SendResult中的offsetMsgId、msgId都可以，返回单条消息
MessageExt viewMessage(String topic,String msgId)
//在指定topic下，根据key进⾏查询，并指定最⼤返回条数，以及开始和结束时间
QueryResult queryMessage(final String topic, final String key,
final int maxNum, final long begin,final long end)
...
}
在内部，实际上都是基于MQAdminImpl这个类来完成的。
viewMessage⽅法：
两种viewMessage⽅法重载形式，都只会返回单条消息。
//初始化Producer
DefaultMQProducer producer = new
DefaultMQProducer("please_rename_unique_group_name");
// Specify name server addresses.
producer.setNamesrvAddr("localhost:9876");
//Launch the instance.
producer.start();
//根据UniqueKey查询
String uniqueKey = "0A1427544F4818B4AAC27DD168880000";
MessageExt msg = producer.viewMessage("TopicTest", uniqueKey);
//打印结果：这⾥仅输出Unique Key与offsetMsgId
MessageClientExt msgExt= (MessageClientExt) msg;
System.out.println("Unique Key:"+msgExt.getMsgId()//即UNIQUE_KEY
+"\noffsetMsgId:"+msgExt.getOffsetMsgId());
//Shut down once the producer instance is not longer in use.
输出结果如下：
Unique Key:0A1427544F4818B4AAC27DD168880000
offsetMsgId:0A14275400002A9F00000000001F268E
如果我们把offsetMsgId当做⽅法参数传⼊，也可以查询到相同的结果。这是因为，在⽅法内部实际上是分两步进⾏查询的：
1、 先把参数当做offsetMsgId，即MessageId进⾏查询；
2、 如果失败，再尝试当做UniqueKey进⾏查询；
源码如下所示：
DefaultMQProducer#viewMessage(String,String)
public MessageExt viewMessage(String topic,
String msgId) throws RemotingException, MQBrokerException,
InterruptedException, MQClientException {
try {
//1 尝试当做offsetMsgId进⾏查询
MessageId oldMsgId = MessageDecoder.decodeMessageId(msgId);
return this.viewMessage(msgId);
} catch (Exception e) {
}
//2 尝试当做UNIQ_KEY进⾏查询
return
this.defaultMQProducerImpl.queryMessageByUniqKey(withNamespace(topic), msgId);
}
前⾯提到，Unique Key只是从逻辑上代表⼀条消息，实际上在Broker端可能存储了多条，因此在当做Unique Key进⾏查询时，会进⾏过滤，只取其中⼀条。
源码如下所示：
MQAdminImpl#queryMessageByUniqKey
public MessageExt queryMessageByUniqKey(String topic,
String uniqKey) throws InterruptedException, MQClientException {
//根据uniqKey进⾏查询
QueryResult qr = this.queryMessage(topic, uniqKey, 32,
MessageClientIDSetter.getNearlyTimeFromID(uniqKey).getTime() -
1000, Long.MAX_VALUE, true);
//对查询结果进⾏过滤，最多只取⼀条
if (qr != null && qr.getMessageList() != null &&
qr.getMessageList().size() > 0) {
return qr.getMessageList().get(0);
} else {
return null;
}
}
实现原理
Unqiue Key & Message Key都需要利⽤RocketMQ的哈希索引机制来完成消息查询，Message Id是在Broker端⽣成的，其包含了Broker地址和commit Log offset信息，可以精确匹配⼀条消息，查询消息更好。
Unique Key 是⽣产者发送消息之前，由RocketMQ 客户端⾃动⽣成的。
DefaultMQProducerImpl#sendKernelImpl
private SendResult sendKernelImpl(final Message msg,
final MessageQueue mq,
final CommunicationMode communicationMode,
final SendCallback sendCallback,
final TopicPublishInfo topicPublishInfo,
final long timeout) {
//省略异常声明
//...略
try {
//如果不是批量消息，则⽣成Unique Key
if (!(msg instanceof MessageBatch)) {
// 设置唯⼀编号
MessageClientIDSetter.setUniqID(msg);
}
//...略
MessageClientIDSetter#setUniqID
public static void setUniqID(final Message msg) {
// Unique Key为空的情况下，才进⾏设置
if
(msg.getProperty(MessageConst.PROPERTY_UNIQ_CLIENT_MESSAGE_ID_KEYIDX) == null)
{
msg.putProperty(MessageConst.PROPERTY_UNIQ_CLIENT_MESSAGE_ID_KEYIDX,
createUniqID());
}
}
Unique Key作⽤
了解Unique Key的作⽤对于我们理解消息重复的原因有很⼤的帮助。RocketMQ并不保证消息投递过程中的Exactly Once语义，即消息只会被精确消费⼀次，需要消费者⾃⼰做幂等。⽽通常导致消息重复消费的原因，主要包括：
- ⽣产者发送时消息重复：
- 消费者Rebalance时消息重复：
导致⽣产者发送重复消息的原因可能是：⼀条消息已被成功发送到服务端并完成持久化，由于⽹络超时此时出现了⽹络闪断或者客户端宕机，导致服务端对客户端应答失败，此时⽣产者将再次尝试发送消息。
在重试发送时，sendKernelImpl会被重复调⽤，意味着setUniqID⽅法会被重复调⽤，不过由于setUniqID⽅法实现中进⾏判空处理，因此重复设置Unique Key。在这种情况下，消费者后续会收到两条内容相同并且 Unique Key 也相同的消息(offsetMsgId不同，因为对Broker来说存储了多次)。
那么消费者如何判断，消费重复是因为重复发送还是Rebalance导致的重复消费呢？
消费者实现MessageListener接⼝监听到的消息类型是MessageExt，可以将其强制转换为
MessageClientExt，之后调⽤getMsgId⽅法获取Unique Key，调⽤getOffsetMsgId获得Message Id。
@Override
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt>
msgs,
ConsumeConcurrentlyContext context) {
for (MessageExt msg:msgs){
MessageClientExt mct = (MessageClientExt)msg;
String uniqueKey = mct.getMsgId();
String messageId = mct.getOffsetMsgId();
}
System.out.printf("%s Receive New Messages: %s %n",
Thread.currentThread().getName(), msgs);
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
批量发送模式下的Unique Key
DefaultMQProducer提供了批量发送消息的接⼝：
public SendResult send(Collection<Message> msgs)
在内部，这批消息⾸先会被构建成⼀个MessageBatch对象。在前⾯sendKernelImpl⽅法中我们也看到了，对于MessageBatch对象，并不会设置Unique Key。
这是因为在将批量消息转换成MessageBatch时，已经设置过了。
⼀个批量消息中每条消息Unique Key是相同的？？？？？？
//Instantiate with a producer group name.
DefaultMQProducer producer = new
DefaultMQProducer("please_rename_unique_group_name");
// Specify name server addresses.
producer.setNamesrvAddr("localhost:9876");
//Launch the instance.
producer.start();
//构建批量消息
ArrayList<Message> msgs = new ArrayList<>();
Message msg1 = new Message("TopicTest",("message3").getBytes());
Message msg2 = new Message("TopicTest",("message4").getBytes());
msgs.add(msg1);
msgs.add(msg2);
//发送
SendResult result = producer.send(msgs);
//打印
System.out.println(result);
输出如下所示：
可以看到，此时输出的msgId(即Unique Key)和offsetMsgId都会包含多个值。客户端给批量消息中每条消息设置不同的Unqiue Key，参考DefaultMQProducer#batch()：
private MessageBatch batch(Collection<Message> msgs) throws MQClientException
{
MessageBatch msgBatch;
try {
//1 将消息集合转换为MessageBatch
msgBatch = MessageBatch.generateFromList(msgs);
//2 迭代每个消息，逐⼀设置Unique Key
for (Message message : msgBatch) {
Validators.checkMessage(message, this);
MessageClientIDSetter.setUniqID(message);
}
//3 设置批量消息的消息体
msgBatch.setBody(msgBatch.encode());
} catch (Exception e) {
throw new MQClientException("Failed to initiate the MessageBatch", e);
}
return msgBatch;
}
Message Id
Message Id是在Broker端⽣成的，⽤于唯⼀标识⼀条消息，在根据Message Id查询的情况下，最多只能查询到⼀条消息。
package org.apache.rocketmq.common.message;
import java.net.SocketAddress;
public class MessageId {
private SocketAddress address;
private long offset;
并提供了⼀个MessageDecoder对象来创建或者解码MessageId。
public static String createMessageId(final ByteBuffer input,
final ByteBuffer addr, final long offset)
public static MessageId decodeMessageId(final String msgId)
Broker端在顺序存储消息时，⾸先会通过createMessageId⽅法创建msgId
CommitLog.DefaultAppendMessageCallback#doAppend
..........................
String msgId;
//3 创建msgId
if ((sysflag & MessageSysFlag.STOREHOSTADDRESS_V6_FLAG) == 0) {
msgId = MessageDecoder.createMessageId(this.msgIdMemory,
msgInner.getStoreHostBytes(storeHostHolder), wroteOffset);
} else {
msgId = MessageDecoder.createMessageId(this.msgIdV6Memory,
msgInner.getStoreHostBytes(storeHostHolder), wroteOffset);
}
⽽客户端在根据msgId向Broker查询消息时，⾸先会将通过MessageDecoder的decodeMessageId⽅法，之后直接向这个broker进⾏查询指定位置的消息。
参⻅：MQAdminImpl#viewMessage
public MessageExt viewMessage(
String msgId) throws RemotingException, MQBrokerException,
InterruptedException, MQClientException {
//1 根据msgId解码成MessageId对象
MessageId messageId = null;
try {
messageId = MessageDecoder.decodeMessageId(msgId);
} catch (Exception e) {
throw new MQClientException(ResponseCode.NO_MESSAGE, "query
message by id finished, but no message.");
}
//2 根据MessageId中的Broker地址和commit log offset信息进⾏查询
return
this.mQClientFactory.getMQClientAPIImpl().viewMessage(RemotingUtil.socketAddre
ss2String(messageId.getAddress()),
messageId.getOffset(), timeoutMillis);
}
由于根据Message Id进⾏查询，实际上是直接从特定Broker的CommitLog中的指定位置进⾏查询的，属于精确匹配，并不像⽤户设置的key，或者Unique Key那么样，需要使⽤到哈希索引机制，因此效率很⾼。
总结
- 3种消息查询⽅式：Message Key & Unique Key & Message Id
- 3种消息查询⼯具：命令⾏、管理平台、客户端API，且⽀持将查询到让特定/所有消费者组重新消费
- 屏蔽Unique Key & Message Id区别，很多地⽅⼆者可以通⽤
- Message Key & Unique Key 需要使⽤到哈希索引机制，有额外的索引维护成本
- Message Id由Broker和commit log offset组成，属于精确匹配，查询效率更好