RocketMQ如何获取指定消息-创新互联

小编给大家分享一下RocketMQ如何获取指定消息，相信大部分人都还不怎么了解，因此分享这篇文章给大家参考一下，希望大家阅读完这篇文章后大有收获，下面让我们一起去了解一下吧！

创新互联建站专注于新余网站建设服务及定制，我们拥有丰富的企业做网站经验。 热诚为您提供新余营销型网站建设，新余网站制作、新余网页设计、新余网站官网定制、成都小程序开发服务，打造新余网络公司原创品牌,更为您提供新余网站排名全网营销落地服务。

概要

消息查询是什么？

消息查询就是根据用户提供的msgId从MQ中取出该消息

RocketMQ如果有多个节点如何查询？

问题：RocketMQ分布式结构中，数据分散在各个节点，即便是同一Topic的数据，也未必都在一个broker上。客户端怎么知道数据该去哪个节点上查？

猜想1：逐个访问broker节点查询数据

猜想2：有某种数据中心存在，该中心知道所有消息存储的位置，只要向该中心查询即可得到消息具体位置，进而取得消息内容

实际：

1.消息Id中含有消息所在的broker的地址信息（IP\Port）以及该消息在CommitLog中的偏移量。

2.客户端实现会从msgId字符串中解析出broker地址，向指定broker节查询消息。

问题：CommitLog文件有多个，只有偏移量估计不能确定在哪个文件吧？

实际：单个Broker节点内offset是全局唯一的，不是每个CommitLog文件的偏移量都是从0开始的。单个节点内所有CommitLog文件共用一套偏移量，每个文件的文件名为其第一个消息的偏移量。所以可以根据偏移量和文件名确定CommitLog文件。

源码阅读

0.使用方式

MessageExt  msg = consumer.viewMessage(msgId);

1.消息ID解析

这个了解下就可以了

public class MessageId {
 private SocketAddress address;
 private long offset;

 public MessageId(SocketAddress address, long offset) {
  this.address = address;
  this.offset = offset;
 }

 //get-set
}

//from MQAdminImpl.java
public MessageExt viewMessage(
 String msgId) throws RemotingException, MQBrokerException, InterruptedException, MQClientException {

 MessageId messageId = null;
 try {
  //从msgId字符串中解析出address和offset
  //address = ip:port
  //offset为消息在CommitLog文件中的偏移量
  messageId = MessageDecoder.decodeMessageId(msgId);
 } catch (Exception e) {
  throw new MQClientException(ResponseCode.NO_MESSAGE, "query message by id finished, but no message.");
 }
 return this.mQClientFactory.getMQClientAPIImpl().viewMessage(RemotingUtil.socketAddress2String(messageId.getAddress()),
  messageId.getOffset(), timeoutMillis);
}

//from MessageDecoder.java
public static MessageId decodeMessageId(final String msgId) throws UnknownHostException {
 SocketAddress address;
 long offset;
 //ipv4和ipv6的区别
 //如果msgId总长度超过32字符，则为ipv6
 int ipLength = msgId.length() == 32 ? 4 * 2 : 16 * 2;

 byte[] ip = UtilAll.string2bytes(msgId.substring(0, ipLength));
 byte[] port = UtilAll.string2bytes(msgId.substring(ipLength, ipLength + 8));
 ByteBuffer bb = ByteBuffer.wrap(port);
 int portInt = bb.getInt(0);
 address = new InetSocketAddress(InetAddress.getByAddress(ip), portInt);

 // offset
 byte[] data = UtilAll.string2bytes(msgId.substring(ipLength + 8, ipLength + 8 + 16));
 bb = ByteBuffer.wrap(data);
 offset = bb.getLong(0);

 return new MessageId(address, offset);
}

分享文章：RocketMQ如何获取指定消息-创新互联
当前网址：http://cdkjz.cn/article/dsccee.html