1.发送消息
Rocketmq Client在发送消息的时候,会根据topic首先从本地缓存获取Broker,获取Broker,如果获取不到,就会到Name Server集群中获取
2.消息偏移量
客户端获取消息偏移量(Consume Offset)的时候,也可能会抛出这个异常:
//RemoteBrokerOffsetStore 类
public long readOffset(final MessageQueue mq, final ReadOffsetType type) {if (mq != null) {switch (type) {case MEMORY_FIRST_THEN_STORE:case READ_FROM_MEMORY: {//省略实现逻辑}case READ_FROM_STORE: {//省略long brokerOffset = this.fetchConsumeOffsetFromBroker(mq);//省略}default:break;}}return -1;
}
从上面的代码中可以看到:获取偏移量的方式有 3 种:
-
MEMORY_FIRST_THEN_STORE:先从内存中获取,如果获取不到,再从 Broker 请求;
-
READ_FROM_MEMORY:直接从内存中获取;
-
READ_FROM_STORE:直接从 Broker 请求。
从Broker获取偏移量
private long fetchConsumeOffsetFromBroker(MessageQueue mq) throws RemotingException, MQBrokerException,
InterruptedException, MQClientException {FindBrokerResult findBrokerResult = this.mQClientFactory.findBrokerAddressInSubscribe(mq.getBrokerName(), MixAll.MASTER_ID, true);if (null == findBrokerResult) {this.mQClientFactory.updateTopicRouteInfoFromNameServer(mq.getTopic());findBrokerResult = this.mQClientFactory.findBrokerAddressInSubscribe(mq.getBrokerName(), MixAll.MASTER_ID, false);}if (findBrokerResult != null) {//忽略处理逻辑} else {throw new MQClientException("The broker[" + mq.getBrokerName() + "] not exist", null);}
}
这段代码跟上一节发送消息时获取 Broker 地址的代码一样,首先从本地内存中获取,如果过去不到,就从 Name Server 中获取,如果取不到,就抛出 Broker 不存在的异常。
3.获取最早消息的保存时间
public long earliestMsgStoreTime(MessageQueue mq) throws MQClientException {String brokerAddr = this.mQClientFactory.findBrokerAddressInPublish(mq.getBrokerName());if (null == brokerAddr) {this.mQClientFactory.updateTopicRouteInfoFromNameServer(mq.getTopic());brokerAddr = this.mQClientFactory.findBrokerAddressInPublish(mq.getBrokerName());}//省略处理逻辑throw new MQClientException("The broker[" + mq.getBrokerName() + "] not exist", null);
}
4.其他获取偏移量方法
除了上面的获取偏移量的方法外,还有 3 个获取偏移量的方法,在 MQAdminImpl 类:
-
searchOffset:从 Broker 获取 Message-Queue 偏移量,跟上面方法类似;
-
maxOffset:从 Broker 获取 MessageQ-ueue 最大偏移量;
-
minOffset:从 Broker 获取 MessageQu-eue 最小偏移量。
5.拉取消息
户端从 Broker 拉取消息之前,首先会从本地缓存获取 Broker 地址,如果获取不到,就从 Name Server 获取 Broker 地址,如果获取失败,则抛出 Broker 不存在的异常。
6.偏移量不合法
如果拉取消息时返回偏移量不合法(OFFSET_ILLEGAL),这时就需要重新处理偏移量。客户端代码的调用关系如下:
这个发生在事务消息的场景,RocketMQ client 向 Broker 拉取消息时,如果 Broker 返回 PULL_OFFSET_MOVED,client 就会通过异步线程(定时 10s 后执行)通知 Broker 更新 offset 为 nextPullOffset(上次 pull 消息时 broker 返回)。代码如下:
public void updateConsumeOffsetToBroker(MessageQueue mq, long offset, boolean isOneway) throws RemotingException,
MQBrokerException, InterruptedException, MQClientException {FindBrokerResult findBrokerResult = this.mQClientFactory.findBrokerAddressInSubscribe(mq.getBrokerName(), MixAll.MASTER_ID, true);if (null == findBrokerResult) {this.mQClientFactory.updateTopicRouteInfoFromNameServer(mq.getTopic());findBrokerResult = this.mQClientFactory.findBrokerAddressInSubscribe(mq.getBrokerName(), MixAll.MASTER_ID, false);}if (findBrokerResult != null) {//省略业务代码} else {throw new MQClientException("The broker[" + mq.getBrokerName() + "] not exist", null);}
}
总结
-
Broker 挂了,客户端定时任务会判断到 Broker 离线,就会从本地缓存中移除(MQClientInstance#cleanOfflineBroker);
-
Broker 网络异常;
-
Broker 发生了主备切换,客户端获取 Broker 地址时切换还没有完