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://scyanting.com/article/dsccee.html